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+"::Dl-Limit",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    closedir(D);
 448    if (chdir(StartDir.c_str()) != 0)
 449       return _error->Errno("chdir",_("Unable to change to %s"),StartDir.c_str());
 450    return true;
 451 }
 452                                                                         /*}}}*/
 453 // Acquire::TotalNeeded - Number of bytes to fetch                      /*{{{*/
 454 // ---------------------------------------------------------------------
 455 /* This is the total number of bytes needed */
 456 double pkgAcquire::TotalNeeded()
 457 {
 458    double Total = 0;
 459    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 460       Total += (*I)->FileSize;
 461    return Total;
 462 }
 463                                                                         /*}}}*/
 464 // Acquire::FetchNeeded - Number of bytes needed to get                 /*{{{*/
 465 // ---------------------------------------------------------------------
 466 /* This is the number of bytes that is not local */
 467 double pkgAcquire::FetchNeeded()
 468 {
 469    double Total = 0;
 470    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 471       if ((*I)->Local == false)
 472          Total += (*I)->FileSize;
 473    return Total;
 474 }
 475                                                                         /*}}}*/
 476 // Acquire::PartialPresent - Number of partial bytes we already have    /*{{{*/
 477 // ---------------------------------------------------------------------
 478 /* This is the number of bytes that is not local */
 479 double pkgAcquire::PartialPresent()
 480 {
 481   double Total = 0;
 482    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 483       if ((*I)->Local == false)
 484          Total += (*I)->PartialSize;
 485    return Total;
 486 }
 487                                                                         /*}}}*/
 488 // Acquire::UriBegin - Start iterator for the uri list                  /*{{{*/
 489 // ---------------------------------------------------------------------
 490 /* */
 491 pkgAcquire::UriIterator pkgAcquire::UriBegin()
 492 {
 493    return UriIterator(Queues);
 494 }
 495                                                                         /*}}}*/
 496 // Acquire::UriEnd - End iterator for the uri list                      /*{{{*/
 497 // ---------------------------------------------------------------------
 498 /* */
 499 pkgAcquire::UriIterator pkgAcquire::UriEnd()
 500 {
 501    return UriIterator(0);
 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 // 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 bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
 549 {
 550    QItem **I = &Items;
 551    // move to the end of the queue and check for duplicates here
 552    for (; *I != 0; I = &(*I)->Next)
 553       if (Item.URI == (*I)->URI)
 554       {
 555          Item.Owner->Status = Item::StatDone;
 556          return false;
 557       }
 558
 559    // Create a new item
 560    QItem *Itm = new QItem;
 561    *Itm = Item;
 562    Itm->Next = 0;
 563    *I = Itm;
 564
 565    Item.Owner->QueueCounter++;
 566    if (Items->Next == 0)
 567       Cycle();
 568    return true;
 569 }
 570                                                                         /*}}}*/
 571 // Queue::Dequeue - Remove an item from the queue                       /*{{{*/
 572 // ---------------------------------------------------------------------
 573 /* We return true if we hit something */
 574 bool pkgAcquire::Queue::Dequeue(Item *Owner)
 575 {
 576    if (Owner->Status == pkgAcquire::Item::StatFetching)
 577       return _error->Error("Tried to dequeue a fetching object");
 578
 579    bool Res = false;
 580
 581    QItem **I = &Items;
 582    for (; *I != 0;)
 583    {
 584       if ((*I)->Owner == Owner)
 585       {
 586          QItem *Jnk= *I;
 587          *I = (*I)->Next;
 588          Owner->QueueCounter--;
 589          delete Jnk;
 590          Res = true;
 591       }
 592       else
 593          I = &(*I)->Next;
 594    }
 595
 596    return Res;
 597 }
 598                                                                         /*}}}*/
 599 // Queue::Startup - Start the worker processes                          /*{{{*/
 600 // ---------------------------------------------------------------------
 601 /* It is possible for this to be called with a pre-existing set of
 602    workers. */
 603 bool pkgAcquire::Queue::Startup()
 604 {
 605    if (Workers == 0)
 606    {
 607       URI U(Name);
 608       pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access);
 609       if (Cnf == 0)
 610          return false;
 611
 612       Workers = new Worker(this,Cnf,Owner->Log);
 613       Owner->Add(Workers);
 614       if (Workers->Start() == false)
 615          return false;
 616
 617       /* When pipelining we commit 10 items. This needs to change when we
 618          added other source retry to have cycle maintain a pipeline depth
 619          on its own. */
 620       if (Cnf->Pipeline == true)
 621          MaxPipeDepth = 1000;
 622       else
 623          MaxPipeDepth = 1;
 624    }
 625
 626    return Cycle();
 627 }
 628                                                                         /*}}}*/
 629 // Queue::Shutdown - Shutdown the worker processes                      /*{{{*/
 630 // ---------------------------------------------------------------------
 631 /* If final is true then all workers are eliminated, otherwise only workers
 632    that do not need cleanup are removed */
 633 bool pkgAcquire::Queue::Shutdown(bool Final)
 634 {
 635    // Delete all of the workers
 636    pkgAcquire::Worker **Cur = &Workers;
 637    while (*Cur != 0)
 638    {
 639       pkgAcquire::Worker *Jnk = *Cur;
 640       if (Final == true || Jnk->GetConf()->NeedsCleanup == false)
 641       {
 642          *Cur = Jnk->NextQueue;
 643          Owner->Remove(Jnk);
 644          delete Jnk;
 645       }
 646       else
 647          Cur = &(*Cur)->NextQueue;
 648    }
 649
 650    return true;
 651 }
 652                                                                         /*}}}*/
 653 // Queue::FindItem - Find a URI in the item list                        /*{{{*/
 654 // ---------------------------------------------------------------------
 655 /* */
 656 pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
 657 {
 658    for (QItem *I = Items; I != 0; I = I->Next)
 659       if (I->URI == URI && I->Worker == Owner)
 660          return I;
 661    return 0;
 662 }
 663                                                                         /*}}}*/
 664 // Queue::ItemDone - Item has been completed                            /*{{{*/
 665 // ---------------------------------------------------------------------
 666 /* The worker signals this which causes the item to be removed from the
 667    queue. If this is the last queue instance then it is removed from the
 668    main queue too.*/
 669 bool pkgAcquire::Queue::ItemDone(QItem *Itm)
 670 {
 671    PipeDepth--;
 672    if (Itm->Owner->Status == pkgAcquire::Item::StatFetching)
 673       Itm->Owner->Status = pkgAcquire::Item::StatDone;
 674
 675    if (Itm->Owner->QueueCounter <= 1)
 676       Owner->Dequeue(Itm->Owner);
 677    else
 678    {
 679       Dequeue(Itm->Owner);
 680       Owner->Bump();
 681    }
 682
 683    return Cycle();
 684 }
 685                                                                         /*}}}*/
 686 // Queue::Cycle - Queue new items into the method                       /*{{{*/
 687 // ---------------------------------------------------------------------
 688 /* This locates a new idle item and sends it to the worker. If pipelining
 689    is enabled then it keeps the pipe full. */
 690 bool pkgAcquire::Queue::Cycle()
 691 {
 692    if (Items == 0 || Workers == 0)
 693       return true;
 694
 695    if (PipeDepth < 0)
 696       return _error->Error("Pipedepth failure");
 697
 698    // Look for a queable item
 699    QItem *I = Items;
 700    while (PipeDepth < (signed)MaxPipeDepth)
 701    {
 702       for (; I != 0; I = I->Next)
 703          if (I->Owner->Status == pkgAcquire::Item::StatIdle)
 704             break;
 705
 706       // Nothing to do, queue is idle.
 707       if (I == 0)
 708          return true;
 709
 710       I->Worker = Workers;
 711       I->Owner->Status = pkgAcquire::Item::StatFetching;
 712       PipeDepth++;
 713       if (Workers->QueueItem(I) == false)
 714          return false;
 715    }
 716
 717    return true;
 718 }
 719                                                                         /*}}}*/
 720 // Queue::Bump - Fetch any pending objects if we are idle               /*{{{*/
 721 // ---------------------------------------------------------------------
 722 /* This is called when an item in multiple queues is dequeued */
 723 void pkgAcquire::Queue::Bump()
 724 {
 725    Cycle();
 726 }
 727                                                                         /*}}}*/
 728 // AcquireStatus::pkgAcquireStatus - Constructor                        /*{{{*/
 729 // ---------------------------------------------------------------------
 730 /* */
 731 pkgAcquireStatus::pkgAcquireStatus() : Update(true), MorePulses(false)
 732 {
 733    Start();
 734 }
 735                                                                         /*}}}*/
 736 // AcquireStatus::Pulse - Called periodically                           /*{{{*/
 737 // ---------------------------------------------------------------------
 738 /* This computes some internal state variables for the derived classes to
 739    use. It generates the current downloaded bytes and total bytes to download
 740    as well as the current CPS estimate. */
 741 bool pkgAcquireStatus::Pulse(pkgAcquire *Owner)
 742 {
 743    TotalBytes = 0;
 744    CurrentBytes = 0;
 745    TotalItems = 0;
 746    CurrentItems = 0;
 747
 748    // Compute the total number of bytes to fetch
 749    unsigned int Unknown = 0;
 750    unsigned int Count = 0;
 751    for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin(); I != Owner->ItemsEnd();
 752         I++, Count++)
 753    {
 754       TotalItems++;
 755       if ((*I)->Status == pkgAcquire::Item::StatDone)
 756          CurrentItems++;
 757
 758       // Totally ignore local items
 759       if ((*I)->Local == true)
 760          continue;
 761
 762       TotalBytes += (*I)->FileSize;
 763       if ((*I)->Complete == true)
 764          CurrentBytes += (*I)->FileSize;
 765       if ((*I)->FileSize == 0 && (*I)->Complete == false)
 766          Unknown++;
 767    }
 768
 769    // Compute the current completion
 770    unsigned long ResumeSize = 0;
 771    for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0;
 772         I = Owner->WorkerStep(I))
 773       if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false)
 774       {
 775          CurrentBytes += I->CurrentSize;
 776          ResumeSize += I->ResumePoint;
 777
 778          // Files with unknown size always have 100% completion
 779          if (I->CurrentItem->Owner->FileSize == 0 &&
 780              I->CurrentItem->Owner->Complete == false)
 781             TotalBytes += I->CurrentSize;
 782       }
 783
 784    // Normalize the figures and account for unknown size downloads
 785    if (TotalBytes <= 0)
 786       TotalBytes = 1;
 787    if (Unknown == Count)
 788       TotalBytes = Unknown;
 789
 790    // Wha?! Is not supposed to happen.
 791    if (CurrentBytes > TotalBytes)
 792       CurrentBytes = TotalBytes;
 793
 794    // Compute the CPS
 795    struct timeval NewTime;
 796    gettimeofday(&NewTime,0);
 797    if ((NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec) ||
 798        NewTime.tv_sec - Time.tv_sec > 6)
 799    {
 800       double Delta = NewTime.tv_sec - Time.tv_sec +
 801                      (NewTime.tv_usec - Time.tv_usec)/1000000.0;
 802
 803       // Compute the CPS value
 804       if (Delta < 0.01)
 805          CurrentCPS = 0;
 806       else
 807          CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta;
 808       LastBytes = CurrentBytes - ResumeSize;
 809       ElapsedTime = (unsigned long)Delta;
 810       Time = NewTime;
 811    }
 812
 813    int fd = _config->FindI("APT::Status-Fd",-1);
 814    if(fd > 0)
 815    {
 816       ostringstream status;
 817
 818       char msg[200];
 819       long i = CurrentItems < TotalItems ? CurrentItems + 1 : CurrentItems;
 820       unsigned long ETA =
 821          (unsigned long)((TotalBytes - CurrentBytes) / CurrentCPS);
 822
 823       // only show the ETA if it makes sense
 824       if (ETA > 0 && ETA < 172800 /* two days */ )
 825          snprintf(msg,sizeof(msg), _("Retrieving file %li of %li (%s remaining)"), i, TotalItems, TimeToStr(ETA).c_str());
 826       else
 827          snprintf(msg,sizeof(msg), _("Retrieving file %li of %li"), i, TotalItems);
 828
 829
 830
 831       // build the status str
 832       status << "dlstatus:" << i
 833              << ":"  << (CurrentBytes/float(TotalBytes)*100.0)
 834              << ":" << msg
 835              << endl;
 836       write(fd, status.str().c_str(), status.str().size());
 837    }
 838
 839    return true;
 840 }
 841                                                                         /*}}}*/
 842 // AcquireStatus::Start - Called when the download is started           /*{{{*/
 843 // ---------------------------------------------------------------------
 844 /* We just reset the counters */
 845 void pkgAcquireStatus::Start()
 846 {
 847    gettimeofday(&Time,0);
 848    gettimeofday(&StartTime,0);
 849    LastBytes = 0;
 850    CurrentCPS = 0;
 851    CurrentBytes = 0;
 852    TotalBytes = 0;
 853    FetchedBytes = 0;
 854    ElapsedTime = 0;
 855    TotalItems = 0;
 856    CurrentItems = 0;
 857 }
 858                                                                         /*}}}*/
 859 // AcquireStatus::Stop - Finished downloading                           /*{{{*/
 860 // ---------------------------------------------------------------------
 861 /* This accurately computes the elapsed time and the total overall CPS. */
 862 void pkgAcquireStatus::Stop()
 863 {
 864    // Compute the CPS and elapsed time
 865    struct timeval NewTime;
 866    gettimeofday(&NewTime,0);
 867
 868    double Delta = NewTime.tv_sec - StartTime.tv_sec +
 869                   (NewTime.tv_usec - StartTime.tv_usec)/1000000.0;
 870
 871    // Compute the CPS value
 872    if (Delta < 0.01)
 873       CurrentCPS = 0;
 874    else
 875       CurrentCPS = FetchedBytes/Delta;
 876    LastBytes = CurrentBytes;
 877    ElapsedTime = (unsigned int)Delta;
 878 }
 879                                                                         /*}}}*/
 880 // AcquireStatus::Fetched - Called when a byte set has been fetched     /*{{{*/
 881 // ---------------------------------------------------------------------
 882 /* This is used to get accurate final transfer rate reporting. */
 883 void pkgAcquireStatus::Fetched(unsigned long Size,unsigned long Resume)
 884 {
 885    FetchedBytes += Size - Resume;
 886 }
 887                                                                         /*}}}*/