X-Git-Url: https://git.saurik.com/apt.git/blobdiff_plain/a656821931a863ffcf46a719a57206eb0d2b11b3..fc5aece9d8600b68cf8d654b379b6d840f2a5524:/apt-pkg/acquire.cc diff --git a/apt-pkg/acquire.cc b/apt-pkg/acquire.cc index 9a546c7e2..70dce4f54 100644 --- a/apt-pkg/acquire.cc +++ b/apt-pkg/acquire.cc @@ -1,6 +1,6 @@ // -*- mode: cpp; mode: fold -*- // Description /*{{{*/ -// $Id: acquire.cc,v 1.15 1998/11/13 07:08:54 jgg Exp $ +// $Id: acquire.cc,v 1.50 2004/03/17 05:17:11 mdz Exp $ /* ###################################################################### Acquire - File Acquiration @@ -21,12 +21,20 @@ #include #include #include -#include +#include +#include + +#include + #include #include +#include +#include /*}}}*/ +using namespace std; + // Acquire::pkgAcquire - Constructor /*{{{*/ // --------------------------------------------------------------------- /* We grab some runtime state from the configuration space */ @@ -45,6 +53,17 @@ pkgAcquire::pkgAcquire(pkgAcquireStatus *Log) : Log(Log) QueueMode = QueueAccess; Debug = _config->FindB("Debug::pkgAcquire",false); + + // This is really a stupid place for this + struct stat St; + if (stat((_config->FindDir("Dir::State::lists") + "partial/").c_str(),&St) != 0 || + S_ISDIR(St.st_mode) == 0) + _error->Error(_("Lists directory %spartial is missing."), + _config->FindDir("Dir::State::lists").c_str()); + if (stat((_config->FindDir("Dir::Cache::Archives") + "partial/").c_str(),&St) != 0 || + S_ISDIR(St.st_mode) == 0) + _error->Error(_("Archive directory %spartial is missing."), + _config->FindDir("Dir::Cache::Archives").c_str()); } /*}}}*/ // Acquire::~pkgAcquire - Destructor /*{{{*/ @@ -52,15 +71,27 @@ pkgAcquire::pkgAcquire(pkgAcquireStatus *Log) : Log(Log) /* Free our memory, clean up the queues (destroy the workers) */ pkgAcquire::~pkgAcquire() { - while (Items.size() != 0) - delete Items[0]; - + Shutdown(); + while (Configs != 0) { MethodConfig *Jnk = Configs; Configs = Configs->Next; delete Jnk; } +} + /*}}}*/ +// Acquire::Shutdown - Clean out the acquire object /*{{{*/ +// --------------------------------------------------------------------- +/* */ +void pkgAcquire::Shutdown() +{ + while (Items.size() != 0) + { + if (Items[0]->Status == Item::StatFetching) + Items[0]->Status = Item::StatError; + delete Items[0]; + } while (Queues != 0) { @@ -84,10 +115,17 @@ void pkgAcquire::Add(Item *Itm) /* Remove an item from the acquire list. This is usually not used.. */ void pkgAcquire::Remove(Item *Itm) { - for (vector::iterator I = Items.begin(); I < Items.end(); I++) + Dequeue(Itm); + + for (ItemIterator I = Items.begin(); I != Items.end();) { if (*I == Itm) + { Items.erase(I); + I = Items.begin(); + } + else + I++; } } /*}}}*/ @@ -124,13 +162,14 @@ void pkgAcquire::Remove(Worker *Work) // Acquire::Enqueue - Queue an URI for fetching /*{{{*/ // --------------------------------------------------------------------- /* This is the entry point for an item. An item calls this function when - it is construction which creates a queue (based on the current queue + it is constructed which creates a queue (based on the current queue mode) and puts the item in that queue. If the system is running then the queue might be started. */ void pkgAcquire::Enqueue(ItemDesc &Item) { // Determine which queue to put the item in - string Name = QueueName(Item.URI); + const MethodConfig *Config; + string Name = QueueName(Item.URI,Config); if (Name.empty() == true) return; @@ -147,6 +186,9 @@ void pkgAcquire::Enqueue(ItemDesc &Item) I->Startup(); } + // See if this is a local only URI + if (Config->LocalOnly == true && Item.Owner->Complete == false) + Item.Owner->Local = true; Item.Owner->Status = Item::StatIdle; // Queue it into the named queue @@ -158,7 +200,7 @@ void pkgAcquire::Enqueue(ItemDesc &Item) { clog << "Fetching " << Item.URI << endl; clog << " to " << Item.Owner->DestFile << endl; - clog << " Queue is: " << QueueName(Item.URI) << endl; + clog << " Queue is: " << Name << endl; } } /*}}}*/ @@ -184,11 +226,11 @@ void pkgAcquire::Dequeue(Item *Itm) /* The string returned depends on the configuration settings and the method parameters. Given something like http://foo.org/bar it can return http://foo.org or http */ -string pkgAcquire::QueueName(string Uri) +string pkgAcquire::QueueName(string Uri,MethodConfig const *&Config) { URI U(Uri); - const MethodConfig *Config = GetConfig(U.Access); + Config = GetConfig(U.Access); if (Config == 0) return string(); @@ -270,7 +312,7 @@ void pkgAcquire::RunFds(fd_set *RSet,fd_set *WSet) /* This runs the queues. It manages a select loop for all of the Worker tasks. The workers interact with the queues and items to manage the actual fetch. */ -bool pkgAcquire::Run() +pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall) { Running = true; @@ -280,10 +322,12 @@ bool pkgAcquire::Run() if (Log != 0) Log->Start(); + bool WasCancelled = false; + // Run till all things have been acquired struct timeval tv; tv.tv_sec = 0; - tv.tv_usec = 500000; + tv.tv_usec = PulseIntervall; while (ToFetch > 0) { fd_set RFds; @@ -293,7 +337,13 @@ bool pkgAcquire::Run() FD_ZERO(&WFds); SetFds(Highest,&RFds,&WFds); - int Res = select(Highest+1,&RFds,&WFds,0,&tv); + int Res; + do + { + Res = select(Highest+1,&RFds,&WFds,0,&tv); + } + while (Res < 0 && errno == EINTR); + if (Res < 0) { _error->Errno("select","Select has failed"); @@ -307,11 +357,14 @@ bool pkgAcquire::Run() // Timeout, notify the log class if (Res == 0 || (Log != 0 && Log->Update == true)) { - tv.tv_usec = 500000; + tv.tv_usec = PulseIntervall; for (Worker *I = Workers; I != 0; I = I->NextAcquire) I->Pulse(); - if (Log != 0) - Log->Pulse(this); + if (Log != 0 && Log->Pulse(this) == false) + { + WasCancelled = true; + break; + } } } @@ -321,9 +374,17 @@ bool pkgAcquire::Run() // Shut down the acquire bits Running = false; for (Queue *I = Queues; I != 0; I = I->Next) - I->Shutdown(); + I->Shutdown(false); - return !_error->PendingError(); + // Shut down the items + for (ItemIterator I = Items.begin(); I != Items.end(); I++) + (*I)->Finished(); + + if (_error->PendingError()) + return Failed; + if (WasCancelled) + return Cancelled; + return Continue; } /*}}}*/ // Acquire::Bump - Called when an item is dequeued /*{{{*/ @@ -352,13 +413,13 @@ bool pkgAcquire::Clean(string Dir) { DIR *D = opendir(Dir.c_str()); if (D == 0) - return _error->Errno("opendir","Unable to read %s",Dir.c_str()); + return _error->Errno("opendir",_("Unable to read %s"),Dir.c_str()); string StartDir = SafeGetCWD(); if (chdir(Dir.c_str()) != 0) { closedir(D); - return _error->Errno("chdir","Unable to change to ",Dir.c_str()); + return _error->Errno("chdir",_("Unable to change to %s"),Dir.c_str()); } for (struct dirent *Dir = readdir(D); Dir != 0; Dir = readdir(D)) @@ -371,7 +432,7 @@ bool pkgAcquire::Clean(string Dir) continue; // Look in the get list - vector::iterator I = Items.begin(); + ItemCIterator I = Items.begin(); for (; I != Items.end(); I++) if (flNotDir((*I)->DestFile) == Dir->d_name) break; @@ -386,25 +447,13 @@ bool pkgAcquire::Clean(string Dir) return true; } /*}}}*/ -// Acquire::MethodConfig::MethodConfig - Constructor /*{{{*/ -// --------------------------------------------------------------------- -/* */ -pkgAcquire::MethodConfig::MethodConfig() -{ - SingleInstance = false; - PreScan = false; - Pipeline = false; - SendConfig = false; - Next = 0; -} - /*}}}*/ // Acquire::TotalNeeded - Number of bytes to fetch /*{{{*/ // --------------------------------------------------------------------- /* This is the total number of bytes needed */ -unsigned long pkgAcquire::TotalNeeded() +double pkgAcquire::TotalNeeded() { - unsigned long Total = 0; - for (pkgAcquire::Item **I = ItemsBegin(); I != ItemsEnd(); I++) + double Total = 0; + for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++) Total += (*I)->FileSize; return Total; } @@ -412,15 +461,57 @@ unsigned long pkgAcquire::TotalNeeded() // Acquire::FetchNeeded - Number of bytes needed to get /*{{{*/ // --------------------------------------------------------------------- /* This is the number of bytes that is not local */ -unsigned long pkgAcquire::FetchNeeded() +double pkgAcquire::FetchNeeded() { - unsigned long Total = 0; - for (pkgAcquire::Item **I = ItemsBegin(); I != ItemsEnd(); I++) + double Total = 0; + for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++) if ((*I)->Local == false) Total += (*I)->FileSize; return Total; } /*}}}*/ +// Acquire::PartialPresent - Number of partial bytes we already have /*{{{*/ +// --------------------------------------------------------------------- +/* This is the number of bytes that is not local */ +double pkgAcquire::PartialPresent() +{ + double Total = 0; + for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++) + if ((*I)->Local == false) + Total += (*I)->PartialSize; + return Total; +} + +// Acquire::UriBegin - Start iterator for the uri list /*{{{*/ +// --------------------------------------------------------------------- +/* */ +pkgAcquire::UriIterator pkgAcquire::UriBegin() +{ + return UriIterator(Queues); +} + /*}}}*/ +// Acquire::UriEnd - End iterator for the uri list /*{{{*/ +// --------------------------------------------------------------------- +/* */ +pkgAcquire::UriIterator pkgAcquire::UriEnd() +{ + return UriIterator(0); +} + /*}}}*/ + +// Acquire::MethodConfig::MethodConfig - Constructor /*{{{*/ +// --------------------------------------------------------------------- +/* */ +pkgAcquire::MethodConfig::MethodConfig() +{ + SingleInstance = false; + Pipeline = false; + SendConfig = false; + LocalOnly = false; + Removable = false; + Next = 0; +} + /*}}}*/ // Queue::Queue - Constructor /*{{{*/ // --------------------------------------------------------------------- @@ -431,6 +522,8 @@ pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name), Items = 0; Next = 0; Workers = 0; + MaxPipeDepth = 1; + PipeDepth = 0; } /*}}}*/ // Queue::~Queue - Destructor /*{{{*/ @@ -438,7 +531,7 @@ pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name), /* */ pkgAcquire::Queue::~Queue() { - Shutdown(); + Shutdown(true); while (Items != 0) { @@ -453,11 +546,14 @@ pkgAcquire::Queue::~Queue() /* */ void pkgAcquire::Queue::Enqueue(ItemDesc &Item) { + QItem **I = &Items; + for (; *I != 0; I = &(*I)->Next); + // Create a new item - QItem *I = new QItem; - I->Next = Items; - Items = I; - *I = Item; + QItem *Itm = new QItem; + *Itm = Item; + Itm->Next = 0; + *I = Itm; Item.Owner->QueueCounter++; if (Items->Next == 0) @@ -466,9 +562,12 @@ void pkgAcquire::Queue::Enqueue(ItemDesc &Item) /*}}}*/ // Queue::Dequeue - Remove an item from the queue /*{{{*/ // --------------------------------------------------------------------- -/* We return true if we hit something*/ +/* We return true if we hit something */ bool pkgAcquire::Queue::Dequeue(Item *Owner) { + if (Owner->Status == pkgAcquire::Item::StatFetching) + return _error->Error("Tried to dequeue a fetching object"); + bool Res = false; QItem **I = &Items; @@ -491,42 +590,59 @@ bool pkgAcquire::Queue::Dequeue(Item *Owner) /*}}}*/ // Queue::Startup - Start the worker processes /*{{{*/ // --------------------------------------------------------------------- -/* */ +/* It is possible for this to be called with a pre-existing set of + workers. */ bool pkgAcquire::Queue::Startup() { - Shutdown(); - - URI U(Name); - pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access); - if (Cnf == 0) - return false; - - Workers = new Worker(this,Cnf,Owner->Log); - Owner->Add(Workers); - if (Workers->Start() == false) - return false; + if (Workers == 0) + { + URI U(Name); + pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access); + if (Cnf == 0) + return false; + + Workers = new Worker(this,Cnf,Owner->Log); + Owner->Add(Workers); + if (Workers->Start() == false) + return false; + + /* When pipelining we commit 10 items. This needs to change when we + added other source retry to have cycle maintain a pipeline depth + on its own. */ + if (Cnf->Pipeline == true) + MaxPipeDepth = 10; + else + MaxPipeDepth = 1; + } return Cycle(); } /*}}}*/ // Queue::Shutdown - Shutdown the worker processes /*{{{*/ // --------------------------------------------------------------------- -/* */ -bool pkgAcquire::Queue::Shutdown() +/* If final is true then all workers are eliminated, otherwise only workers + that do not need cleanup are removed */ +bool pkgAcquire::Queue::Shutdown(bool Final) { // Delete all of the workers - while (Workers != 0) + pkgAcquire::Worker **Cur = &Workers; + while (*Cur != 0) { - pkgAcquire::Worker *Jnk = Workers; - Workers = Workers->NextQueue; - Owner->Remove(Jnk); - delete Jnk; + pkgAcquire::Worker *Jnk = *Cur; + if (Final == true || Jnk->GetConf()->NeedsCleanup == false) + { + *Cur = Jnk->NextQueue; + Owner->Remove(Jnk); + delete Jnk; + } + else + Cur = &(*Cur)->NextQueue; } return true; } /*}}}*/ -// Queue::Finditem - Find a URI in the item list /*{{{*/ +// Queue::FindItem - Find a URI in the item list /*{{{*/ // --------------------------------------------------------------------- /* */ pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner) @@ -544,6 +660,10 @@ pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Wor main queue too.*/ bool pkgAcquire::Queue::ItemDone(QItem *Itm) { + PipeDepth--; + if (Itm->Owner->Status == pkgAcquire::Item::StatFetching) + Itm->Owner->Status = pkgAcquire::Item::StatDone; + if (Itm->Owner->QueueCounter <= 1) Owner->Dequeue(Itm->Owner); else @@ -557,39 +677,51 @@ bool pkgAcquire::Queue::ItemDone(QItem *Itm) /*}}}*/ // Queue::Cycle - Queue new items into the method /*{{{*/ // --------------------------------------------------------------------- -/* This locates a new idle item and sends it to the worker */ +/* This locates a new idle item and sends it to the worker. If pipelining + is enabled then it keeps the pipe full. */ bool pkgAcquire::Queue::Cycle() { if (Items == 0 || Workers == 0) return true; + if (PipeDepth < 0) + return _error->Error("Pipedepth failure"); + // Look for a queable item QItem *I = Items; - for (; I != 0; I = I->Next) - if (I->Owner->Status == pkgAcquire::Item::StatIdle) - break; - - // Nothing to do, queue is idle. - if (I == 0) - return true; + while (PipeDepth < (signed)MaxPipeDepth) + { + for (; I != 0; I = I->Next) + if (I->Owner->Status == pkgAcquire::Item::StatIdle) + break; + + // Nothing to do, queue is idle. + if (I == 0) + return true; + + I->Worker = Workers; + I->Owner->Status = pkgAcquire::Item::StatFetching; + PipeDepth++; + if (Workers->QueueItem(I) == false) + return false; + } - I->Worker = Workers; - I->Owner->Status = pkgAcquire::Item::StatFetching; - return Workers->QueueItem(I); + return true; } /*}}}*/ // Queue::Bump - Fetch any pending objects if we are idle /*{{{*/ // --------------------------------------------------------------------- -/* */ +/* This is called when an item in multiple queues is dequeued */ void pkgAcquire::Queue::Bump() { + Cycle(); } /*}}}*/ // AcquireStatus::pkgAcquireStatus - Constructor /*{{{*/ // --------------------------------------------------------------------- /* */ -pkgAcquireStatus::pkgAcquireStatus() +pkgAcquireStatus::pkgAcquireStatus() : Update(true), MorePulses(false) { Start(); } @@ -599,21 +731,27 @@ pkgAcquireStatus::pkgAcquireStatus() /* This computes some internal state variables for the derived classes to use. It generates the current downloaded bytes and total bytes to download as well as the current CPS estimate. */ -void pkgAcquireStatus::Pulse(pkgAcquire *Owner) +bool pkgAcquireStatus::Pulse(pkgAcquire *Owner) { TotalBytes = 0; CurrentBytes = 0; + TotalItems = 0; + CurrentItems = 0; // Compute the total number of bytes to fetch unsigned int Unknown = 0; unsigned int Count = 0; - for (pkgAcquire::Item **I = Owner->ItemsBegin(); I != Owner->ItemsEnd(); + for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin(); I != Owner->ItemsEnd(); I++, Count++) { + TotalItems++; + if ((*I)->Status == pkgAcquire::Item::StatDone) + CurrentItems++; + // Totally ignore local items if ((*I)->Local == true) continue; - + TotalBytes += (*I)->FileSize; if ((*I)->Complete == true) CurrentBytes += (*I)->FileSize; @@ -622,18 +760,29 @@ void pkgAcquireStatus::Pulse(pkgAcquire *Owner) } // Compute the current completion + unsigned long ResumeSize = 0; for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0; I = Owner->WorkerStep(I)) if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false) + { CurrentBytes += I->CurrentSize; - + ResumeSize += I->ResumePoint; + + // Files with unknown size always have 100% completion + if (I->CurrentItem->Owner->FileSize == 0 && + I->CurrentItem->Owner->Complete == false) + TotalBytes += I->CurrentSize; + } + // Normalize the figures and account for unknown size downloads if (TotalBytes <= 0) TotalBytes = 1; if (Unknown == Count) TotalBytes = Unknown; - else - TotalBytes += TotalBytes/(Count - Unknown)*Unknown; + + // Wha?! Is not supposed to happen. + if (CurrentBytes > TotalBytes) + CurrentBytes = TotalBytes; // Compute the CPS struct timeval NewTime; @@ -641,23 +790,20 @@ void pkgAcquireStatus::Pulse(pkgAcquire *Owner) if (NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec || NewTime.tv_sec - Time.tv_sec > 6) { - // Compute the delta time with full accuracy - long usdiff = NewTime.tv_usec - Time.tv_usec; - long sdiff = NewTime.tv_sec - Time.tv_sec; + double Delta = NewTime.tv_sec - Time.tv_sec + + (NewTime.tv_usec - Time.tv_usec)/1000000.0; - // Borrow - if (usdiff < 0) - { - usdiff += 1000000; - sdiff--; - } - // Compute the CPS value - CurrentCPS = (CurrentBytes - LastBytes)/(sdiff + usdiff/1000000.0); - LastBytes = CurrentBytes; - ElapsedTime = NewTime.tv_sec - StartTime.tv_sec; + if (Delta < 0.01) + CurrentCPS = 0; + else + CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta; + LastBytes = CurrentBytes - ResumeSize; + ElapsedTime = (unsigned long)Delta; Time = NewTime; } + + return true; } /*}}}*/ // AcquireStatus::Start - Called when the download is started /*{{{*/ @@ -673,6 +819,8 @@ void pkgAcquireStatus::Start() TotalBytes = 0; FetchedBytes = 0; ElapsedTime = 0; + TotalItems = 0; + CurrentItems = 0; } /*}}}*/ // AcquireStatus::Stop - Finished downloading /*{{{*/ @@ -684,21 +832,16 @@ void pkgAcquireStatus::Stop() struct timeval NewTime; gettimeofday(&NewTime,0); - // Compute the delta time with full accuracy - long usdiff = NewTime.tv_usec - StartTime.tv_usec; - long sdiff = NewTime.tv_sec - StartTime.tv_sec; - - // Borrow - if (usdiff < 0) - { - usdiff += 1000000; - sdiff--; - } + double Delta = NewTime.tv_sec - StartTime.tv_sec + + (NewTime.tv_usec - StartTime.tv_usec)/1000000.0; // Compute the CPS value - CurrentCPS = FetchedBytes/(sdiff + usdiff/1000000.0); + if (Delta < 0.01) + CurrentCPS = 0; + else + CurrentCPS = FetchedBytes/Delta; LastBytes = CurrentBytes; - ElapsedTime = sdiff; + ElapsedTime = (unsigned int)Delta; } /*}}}*/ // AcquireStatus::Fetched - Called when a byte set has been fetched /*{{{*/