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