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