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