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