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