X-Git-Url: https://git.hcoop.net/ntk/apt.git/blobdiff_plain/4b65cc13c74231e14a313ea05a2376cbc5155c5f..HEAD:/apt-pkg/acquire.cc diff --git a/apt-pkg/acquire.cc b/apt-pkg/acquire.cc index daea234d..057bc24c 100644 --- a/apt-pkg/acquire.cc +++ b/apt-pkg/acquire.cc @@ -5,30 +5,39 @@ Acquire - File Acquiration - The core element for the schedual system is the concept of a named + The core element for the schedule system is the concept of a named queue. Each queue is unique and each queue has a name derived from the - URI. The degree of paralization can be controled by how the queue + URI. The degree of paralization can be controlled by how the queue name is derived from the URI. ##################################################################### */ /*}}}*/ // Include Files /*{{{*/ +#include + #include #include #include #include #include #include +#include -#include - +#include +#include #include #include - +#include +#include +#include +#include + #include #include +#include #include -#include + +#include /*}}}*/ using namespace std; @@ -36,32 +45,60 @@ using namespace std; // Acquire::pkgAcquire - Constructor /*{{{*/ // --------------------------------------------------------------------- /* We grab some runtime state from the configuration space */ -pkgAcquire::pkgAcquire(pkgAcquireStatus *Log) : Log(Log) +pkgAcquire::pkgAcquire() : LockFD(-1), Queues(0), Workers(0), Configs(0), Log(NULL), ToFetch(0), + Debug(_config->FindB("Debug::pkgAcquire",false)), + Running(false) { - Queues = 0; - Configs = 0; - Workers = 0; - ToFetch = 0; - Running = false; - - string Mode = _config->Find("Acquire::Queue-Mode","host"); + string const Mode = _config->Find("Acquire::Queue-Mode","host"); + if (strcasecmp(Mode.c_str(),"host") == 0) + QueueMode = QueueHost; + if (strcasecmp(Mode.c_str(),"access") == 0) + QueueMode = QueueAccess; +} +pkgAcquire::pkgAcquire(pkgAcquireStatus *Progress) : LockFD(-1), Queues(0), Workers(0), + Configs(0), Log(Progress), ToFetch(0), + Debug(_config->FindB("Debug::pkgAcquire",false)), + Running(false) +{ + string const Mode = _config->Find("Acquire::Queue-Mode","host"); if (strcasecmp(Mode.c_str(),"host") == 0) QueueMode = QueueHost; if (strcasecmp(Mode.c_str(),"access") == 0) - QueueMode = QueueAccess; + QueueMode = QueueAccess; + Setup(Progress, ""); +} + /*}}}*/ +// Acquire::Setup - Delayed Constructor /*{{{*/ +// --------------------------------------------------------------------- +/* Do everything needed to be a complete Acquire object and report the + success (or failure) back so the user knows that something is wrong… */ +bool pkgAcquire::Setup(pkgAcquireStatus *Progress, string const &Lock) +{ + Log = Progress; - 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()); + // check for existence and possibly create auxiliary directories + string const listDir = _config->FindDir("Dir::State::lists"); + string const partialListDir = listDir + "partial/"; + string const archivesDir = _config->FindDir("Dir::Cache::Archives"); + string const partialArchivesDir = archivesDir + "partial/"; + + if (CreateAPTDirectoryIfNeeded(_config->FindDir("Dir::State"), partialListDir) == false && + CreateAPTDirectoryIfNeeded(listDir, partialListDir) == false) + return _error->Errno("Acquire", _("List directory %spartial is missing."), listDir.c_str()); + + if (CreateAPTDirectoryIfNeeded(_config->FindDir("Dir::Cache"), partialArchivesDir) == false && + CreateAPTDirectoryIfNeeded(archivesDir, partialArchivesDir) == false) + return _error->Errno("Acquire", _("Archives directory %spartial is missing."), archivesDir.c_str()); + + if (Lock.empty() == true || _config->FindB("Debug::NoLocking", false) == true) + return true; + + // Lock the directory this acquire object will work in + LockFD = GetLock(flCombine(Lock, "lock")); + if (LockFD == -1) + return _error->Error(_("Unable to lock directory %s"), Lock.c_str()); + + return true; } /*}}}*/ // Acquire::~pkgAcquire - Destructor /*{{{*/ @@ -70,7 +107,10 @@ pkgAcquire::pkgAcquire(pkgAcquireStatus *Log) : Log(Log) pkgAcquire::~pkgAcquire() { Shutdown(); - + + if (LockFD != -1) + close(LockFD); + while (Configs != 0) { MethodConfig *Jnk = Configs; @@ -84,7 +124,7 @@ pkgAcquire::~pkgAcquire() /* */ void pkgAcquire::Shutdown() { - while (Items.size() != 0) + while (Items.empty() == false) { if (Items[0]->Status == Item::StatFetching) Items[0]->Status = Item::StatError; @@ -123,7 +163,7 @@ void pkgAcquire::Remove(Item *Itm) I = Items.begin(); } else - I++; + ++I; } } /*}}}*/ @@ -141,7 +181,7 @@ void pkgAcquire::Add(Worker *Work) // --------------------------------------------------------------------- /* A worker has died. This can not be done while the select loop is running as it would require that RunFds could handling a changing list state and - it cant.. */ + it can't.. */ void pkgAcquire::Remove(Worker *Work) { if (Running == true) @@ -210,11 +250,19 @@ void pkgAcquire::Dequeue(Item *Itm) { Queue *I = Queues; bool Res = false; - for (; I != 0; I = I->Next) - Res |= I->Dequeue(Itm); - if (Debug == true) clog << "Dequeuing " << Itm->DestFile << endl; + + for (; I != 0; I = I->Next) + { + if (I->Dequeue(Itm)) + { + Res = true; + if (Debug == true) + clog << "Dequeued from " << I->Name << endl; + } + } + if (Res == true) ToFetch--; } @@ -235,9 +283,30 @@ string pkgAcquire::QueueName(string Uri,MethodConfig const *&Config) /* Single-Instance methods get exactly one queue per URI. This is also used for the Access queue method */ if (Config->SingleInstance == true || QueueMode == QueueAccess) - return U.Access; + return U.Access; + + string AccessSchema = U.Access + ':', + FullQueueName = AccessSchema + U.Host; + unsigned int Instances = 0, SchemaLength = AccessSchema.length(); + + Queue *I = Queues; + for (; I != 0; I = I->Next) { + // if the queue already exists, re-use it + if (I->Name == FullQueueName) + return FullQueueName; + + if (I->Name.compare(0, SchemaLength, AccessSchema) == 0) + Instances++; + } + + if (Debug) { + clog << "Found " << Instances << " instances of " << U.Access << endl; + } + + if (Instances >= (unsigned int)_config->FindI("Acquire::QueueHost::Limit",10)) + return U.Access; - return U.Access + ':' + U.Host; + return FullQueueName; } /*}}}*/ // Acquire::GetConfig - Fetch the configuration information /*{{{*/ @@ -379,7 +448,7 @@ pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall) I->Shutdown(false); // Shut down the items - for (ItemIterator I = Items.begin(); I != Items.end(); I++) + for (ItemIterator I = Items.begin(); I != Items.end(); ++I) (*I)->Finished(); if (_error->PendingError()) @@ -405,7 +474,7 @@ void pkgAcquire::Bump() pkgAcquire::Worker *pkgAcquire::WorkerStep(Worker *I) { return I->NextAcquire; -}; +} /*}}}*/ // Acquire::Clean - Cleans a directory /*{{{*/ // --------------------------------------------------------------------- @@ -413,6 +482,13 @@ pkgAcquire::Worker *pkgAcquire::WorkerStep(Worker *I) if it is part of the download set. */ bool pkgAcquire::Clean(string Dir) { + // non-existing directories are by definition clean… + if (DirectoryExists(Dir) == false) + return true; + + if(Dir == "/") + return _error->Error(_("Clean of %s is not supported"), Dir.c_str()); + DIR *D = opendir(Dir.c_str()); if (D == 0) return _error->Errno("opendir",_("Unable to read %s"),Dir.c_str()); @@ -435,7 +511,7 @@ bool pkgAcquire::Clean(string Dir) // Look in the get list ItemCIterator I = Items.begin(); - for (; I != Items.end(); I++) + for (; I != Items.end(); ++I) if (flNotDir((*I)->DestFile) == Dir->d_name) break; @@ -453,10 +529,10 @@ bool pkgAcquire::Clean(string Dir) // Acquire::TotalNeeded - Number of bytes to fetch /*{{{*/ // --------------------------------------------------------------------- /* This is the total number of bytes needed */ -double pkgAcquire::TotalNeeded() +APT_PURE unsigned long long pkgAcquire::TotalNeeded() { - double Total = 0; - for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++) + unsigned long long Total = 0; + for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I) Total += (*I)->FileSize; return Total; } @@ -464,10 +540,10 @@ double pkgAcquire::TotalNeeded() // Acquire::FetchNeeded - Number of bytes needed to get /*{{{*/ // --------------------------------------------------------------------- /* This is the number of bytes that is not local */ -double pkgAcquire::FetchNeeded() +APT_PURE unsigned long long pkgAcquire::FetchNeeded() { - double Total = 0; - for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++) + unsigned long long Total = 0; + for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I) if ((*I)->Local == false) Total += (*I)->FileSize; return Total; @@ -476,15 +552,15 @@ double pkgAcquire::FetchNeeded() // Acquire::PartialPresent - Number of partial bytes we already have /*{{{*/ // --------------------------------------------------------------------- /* This is the number of bytes that is not local */ -double pkgAcquire::PartialPresent() +APT_PURE unsigned long long pkgAcquire::PartialPresent() { - double Total = 0; - for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++) + unsigned long long 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 /*{{{*/ // --------------------------------------------------------------------- /* */ @@ -501,7 +577,6 @@ pkgAcquire::UriIterator pkgAcquire::UriEnd() return UriIterator(0); } /*}}}*/ - // Acquire::MethodConfig::MethodConfig - Constructor /*{{{*/ // --------------------------------------------------------------------- /* */ @@ -515,7 +590,6 @@ pkgAcquire::MethodConfig::MethodConfig() Next = 0; } /*}}}*/ - // Queue::Queue - Constructor /*{{{*/ // --------------------------------------------------------------------- /* */ @@ -620,7 +694,7 @@ bool pkgAcquire::Queue::Startup() added other source retry to have cycle maintain a pipeline depth on its own. */ if (Cnf->Pipeline == true) - MaxPipeDepth = 1000; + MaxPipeDepth = _config->FindI("Acquire::Max-Pipeline-Depth",10); else MaxPipeDepth = 1; } @@ -727,11 +801,10 @@ void pkgAcquire::Queue::Bump() Cycle(); } /*}}}*/ - // AcquireStatus::pkgAcquireStatus - Constructor /*{{{*/ // --------------------------------------------------------------------- /* */ -pkgAcquireStatus::pkgAcquireStatus() : Update(true), MorePulses(false) +pkgAcquireStatus::pkgAcquireStatus() : d(NULL), Update(true), MorePulses(false) { Start(); } @@ -752,11 +825,11 @@ bool pkgAcquireStatus::Pulse(pkgAcquire *Owner) unsigned int Unknown = 0; unsigned int Count = 0; for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin(); I != Owner->ItemsEnd(); - I++, Count++) + ++I, ++Count) { TotalItems++; if ((*I)->Status == pkgAcquire::Item::StatDone) - CurrentItems++; + ++CurrentItems; // Totally ignore local items if ((*I)->Local == true) @@ -766,11 +839,11 @@ bool pkgAcquireStatus::Pulse(pkgAcquire *Owner) if ((*I)->Complete == true) CurrentBytes += (*I)->FileSize; if ((*I)->FileSize == 0 && (*I)->Complete == false) - Unknown++; + ++Unknown; } // Compute the current completion - unsigned long ResumeSize = 0; + unsigned long long ResumeSize = 0; for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0; I = Owner->WorkerStep(I)) if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false) @@ -809,7 +882,7 @@ bool pkgAcquireStatus::Pulse(pkgAcquire *Owner) else CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta; LastBytes = CurrentBytes - ResumeSize; - ElapsedTime = (unsigned long)Delta; + ElapsedTime = (unsigned long long)Delta; Time = NewTime; } @@ -820,8 +893,9 @@ bool pkgAcquireStatus::Pulse(pkgAcquire *Owner) char msg[200]; long i = CurrentItems < TotalItems ? CurrentItems + 1 : CurrentItems; - unsigned long ETA = - (unsigned long)((TotalBytes - CurrentBytes) / CurrentCPS); + unsigned long long ETA = 0; + if(CurrentCPS > 0) + ETA = (TotalBytes - CurrentBytes) / CurrentCPS; // only show the ETA if it makes sense if (ETA > 0 && ETA < 172800 /* two days */ ) @@ -836,7 +910,9 @@ bool pkgAcquireStatus::Pulse(pkgAcquire *Owner) << ":" << (CurrentBytes/float(TotalBytes)*100.0) << ":" << msg << endl; - write(fd, status.str().c_str(), status.str().size()); + + std::string const dlstatus = status.str(); + FileFd::Write(fd, dlstatus.c_str(), dlstatus.size()); } return true; @@ -877,13 +953,13 @@ void pkgAcquireStatus::Stop() else CurrentCPS = FetchedBytes/Delta; LastBytes = CurrentBytes; - ElapsedTime = (unsigned int)Delta; + ElapsedTime = (unsigned long long)Delta; } /*}}}*/ // AcquireStatus::Fetched - Called when a byte set has been fetched /*{{{*/ // --------------------------------------------------------------------- /* This is used to get accurate final transfer rate reporting. */ -void pkgAcquireStatus::Fetched(unsigned long Size,unsigned long Resume) +void pkgAcquireStatus::Fetched(unsigned long long Size,unsigned long long Resume) { FetchedBytes += Size - Resume; }