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