+ Pipeline = false;
+ SendConfig = false;
+ LocalOnly = false;
+ Removable = false;
+ Next = 0;
+}
+ /*}}}*/
+// Queue::Queue - Constructor /*{{{*/
+// ---------------------------------------------------------------------
+/* */
+pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name),
+ Owner(Owner)
+{
+ Items = 0;
+ Next = 0;
+ Workers = 0;
+ MaxPipeDepth = 1;
+ PipeDepth = 0;
+}
+ /*}}}*/
+// Queue::~Queue - Destructor /*{{{*/
+// ---------------------------------------------------------------------
+/* */
+pkgAcquire::Queue::~Queue()
+{
+ Shutdown(true);
+
+ while (Items != 0)
+ {
+ QItem *Jnk = Items;
+ Items = Items->Next;
+ delete Jnk;
+ }
+}
+ /*}}}*/
+// Queue::Enqueue - Queue an item to the queue /*{{{*/
+// ---------------------------------------------------------------------
+/* */
+bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
+{
+ QItem **I = &Items;
+ // move to the end of the queue and check for duplicates here
+ for (; *I != 0; I = &(*I)->Next)
+ if (Item.URI == (*I)->URI)
+ {
+ Item.Owner->Status = Item::StatDone;
+ return false;
+ }
+
+ // Create a new item
+ QItem *Itm = new QItem;
+ *Itm = Item;
+ Itm->Next = 0;
+ *I = Itm;
+
+ Item.Owner->QueueCounter++;
+ if (Items->Next == 0)
+ Cycle();
+ return true;
+}
+ /*}}}*/
+// Queue::Dequeue - Remove an item from the queue /*{{{*/
+// ---------------------------------------------------------------------
+/* 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;
+ for (; *I != 0;)
+ {
+ if ((*I)->Owner == Owner)
+ {
+ QItem *Jnk= *I;
+ *I = (*I)->Next;
+ Owner->QueueCounter--;
+ delete Jnk;
+ Res = true;
+ }
+ else
+ I = &(*I)->Next;
+ }
+
+ return Res;
+}
+ /*}}}*/
+// 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()
+{
+ 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 = _config->FindI("Acquire::Max-Pipeline-Depth",10);
+ else
+ MaxPipeDepth = 1;
+ }
+
+ return Cycle();
+}
+ /*}}}*/
+// Queue::Shutdown - Shutdown the worker processes /*{{{*/
+// ---------------------------------------------------------------------
+/* 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
+ pkgAcquire::Worker **Cur = &Workers;
+ while (*Cur != 0)
+ {
+ 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 /*{{{*/
+// ---------------------------------------------------------------------
+/* */
+pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
+{
+ for (QItem *I = Items; I != 0; I = I->Next)
+ if (I->URI == URI && I->Worker == Owner)
+ return I;
+ return 0;
+}
+ /*}}}*/
+// Queue::ItemDone - Item has been completed /*{{{*/
+// ---------------------------------------------------------------------
+/* The worker signals this which causes the item to be removed from the
+ queue. If this is the last queue instance then it is removed from the
+ 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
+ {
+ Dequeue(Itm->Owner);
+ Owner->Bump();
+ }
+
+ return Cycle();
+}
+ /*}}}*/
+// Queue::Cycle - Queue new items into the method /*{{{*/
+// ---------------------------------------------------------------------
+/* 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;
+ 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;
+ }
+
+ 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() : Update(true), MorePulses(false)
+{
+ Start();
+}
+ /*}}}*/
+// AcquireStatus::Pulse - Called periodically /*{{{*/
+// ---------------------------------------------------------------------
+/* 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. */
+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::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;
+ if ((*I)->FileSize == 0 && (*I)->Complete == false)
+ Unknown++;
+ }
+
+ // 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;
+
+ // Wha?! Is not supposed to happen.
+ if (CurrentBytes > TotalBytes)
+ CurrentBytes = TotalBytes;
+
+ // Compute the CPS
+ struct timeval NewTime;
+ gettimeofday(&NewTime,0);
+ if ((NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec) ||
+ NewTime.tv_sec - Time.tv_sec > 6)
+ {
+ double Delta = NewTime.tv_sec - Time.tv_sec +
+ (NewTime.tv_usec - Time.tv_usec)/1000000.0;
+
+ // Compute the CPS value
+ if (Delta < 0.01)
+ CurrentCPS = 0;
+ else
+ CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta;
+ LastBytes = CurrentBytes - ResumeSize;
+ ElapsedTime = (unsigned long)Delta;
+ Time = NewTime;
+ }
+
+ int fd = _config->FindI("APT::Status-Fd",-1);
+ if(fd > 0)
+ {
+ ostringstream status;
+
+ char msg[200];
+ long i = CurrentItems < TotalItems ? CurrentItems + 1 : CurrentItems;
+ unsigned long ETA =
+ (unsigned long)((TotalBytes - CurrentBytes) / CurrentCPS);
+
+ // only show the ETA if it makes sense
+ if (ETA > 0 && ETA < 172800 /* two days */ )
+ snprintf(msg,sizeof(msg), _("Retrieving file %li of %li (%s remaining)"), i, TotalItems, TimeToStr(ETA).c_str());
+ else
+ snprintf(msg,sizeof(msg), _("Retrieving file %li of %li"), i, TotalItems);
+
+
+
+ // build the status str
+ status << "dlstatus:" << i
+ << ":" << (CurrentBytes/float(TotalBytes)*100.0)
+ << ":" << msg
+ << endl;
+ write(fd, status.str().c_str(), status.str().size());
+ }
+
+ return true;
+}
+ /*}}}*/
+// AcquireStatus::Start - Called when the download is started /*{{{*/
+// ---------------------------------------------------------------------
+/* We just reset the counters */
+void pkgAcquireStatus::Start()
+{
+ gettimeofday(&Time,0);
+ gettimeofday(&StartTime,0);
+ LastBytes = 0;
+ CurrentCPS = 0;
+ CurrentBytes = 0;
+ TotalBytes = 0;
+ FetchedBytes = 0;
+ ElapsedTime = 0;
+ TotalItems = 0;
+ CurrentItems = 0;
+}
+ /*}}}*/
+// AcquireStatus::Stop - Finished downloading /*{{{*/
+// ---------------------------------------------------------------------
+/* This accurately computes the elapsed time and the total overall CPS. */
+void pkgAcquireStatus::Stop()
+{
+ // Compute the CPS and elapsed time
+ struct timeval NewTime;
+ gettimeofday(&NewTime,0);
+
+ double Delta = NewTime.tv_sec - StartTime.tv_sec +
+ (NewTime.tv_usec - StartTime.tv_usec)/1000000.0;
+
+ // Compute the CPS value
+ if (Delta < 0.01)
+ CurrentCPS = 0;
+ else
+ CurrentCPS = FetchedBytes/Delta;
+ LastBytes = CurrentBytes;
+ ElapsedTime = (unsigned int)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)
+{
+ FetchedBytes += Size - Resume;
projects / apt.git / blobdiff