]> git.saurik.com Git - apt.git/blame_incremental - apt-pkg/acquire.cc
merged from lp:~donkult/apt/sid
[apt.git] / apt-pkg / acquire.cc
... / ...
CommitLineData
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 schedual 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 controled by how the queue
11 name is derived from the URI.
12
13 ##################################################################### */
14 /*}}}*/
15// Include Files /*{{{*/
16#include <apt-pkg/acquire.h>
17#include <apt-pkg/acquire-item.h>
18#include <apt-pkg/acquire-worker.h>
19#include <apt-pkg/configuration.h>
20#include <apt-pkg/error.h>
21#include <apt-pkg/strutl.h>
22#include <apt-pkg/fileutl.h>
23
24#include <apti18n.h>
25
26#include <iostream>
27#include <sstream>
28#include <stdio.h>
29
30#include <dirent.h>
31#include <sys/time.h>
32#include <errno.h>
33 /*}}}*/
34
35using namespace std;
36
37// Acquire::pkgAcquire - Constructor /*{{{*/
38// ---------------------------------------------------------------------
39/* We grab some runtime state from the configuration space */
40pkgAcquire::pkgAcquire() : Queues(0), Workers(0), Configs(0), Log(NULL), ToFetch(0),
41 Debug(_config->FindB("Debug::pkgAcquire",false)),
42 Running(false), LockFD(-1)
43{
44 string const Mode = _config->Find("Acquire::Queue-Mode","host");
45 if (strcasecmp(Mode.c_str(),"host") == 0)
46 QueueMode = QueueHost;
47 if (strcasecmp(Mode.c_str(),"access") == 0)
48 QueueMode = QueueAccess;
49}
50pkgAcquire::pkgAcquire(pkgAcquireStatus *Progress) : Queues(0), Workers(0),
51 Configs(0), Log(Progress), ToFetch(0),
52 Debug(_config->FindB("Debug::pkgAcquire",false)),
53 Running(false), LockFD(-1)
54{
55 string const Mode = _config->Find("Acquire::Queue-Mode","host");
56 if (strcasecmp(Mode.c_str(),"host") == 0)
57 QueueMode = QueueHost;
58 if (strcasecmp(Mode.c_str(),"access") == 0)
59 QueueMode = QueueAccess;
60 Setup(Progress, "");
61}
62 /*}}}*/
63// Acquire::Setup - Delayed Constructor /*{{{*/
64// ---------------------------------------------------------------------
65/* Do everything needed to be a complete Acquire object and report the
66 success (or failure) back so the user knows that something is wrong… */
67bool pkgAcquire::Setup(pkgAcquireStatus *Progress, string const &Lock)
68{
69 Log = Progress;
70
71 // check for existence and possibly create auxiliary directories
72 string const listDir = _config->FindDir("Dir::State::lists");
73 string const partialListDir = listDir + "partial/";
74 string const archivesDir = _config->FindDir("Dir::Cache::Archives");
75 string const partialArchivesDir = archivesDir + "partial/";
76
77 if (CreateAPTDirectoryIfNeeded(_config->FindDir("Dir::State"), partialListDir) == false &&
78 CreateAPTDirectoryIfNeeded(listDir, partialListDir) == false)
79 return _error->Errno("Acquire", _("List directory %spartial is missing."), listDir.c_str());
80
81 if (CreateAPTDirectoryIfNeeded(_config->FindDir("Dir::Cache"), partialArchivesDir) == false &&
82 CreateAPTDirectoryIfNeeded(archivesDir, partialArchivesDir) == false)
83 return _error->Errno("Acquire", _("Archives directory %spartial is missing."), archivesDir.c_str());
84
85 if (Lock.empty() == true || _config->FindB("Debug::NoLocking", false) == true)
86 return true;
87
88 // Lock the directory this acquire object will work in
89 LockFD = GetLock(flCombine(Lock, "lock"));
90 if (LockFD == -1)
91 return _error->Error(_("Unable to lock directory %s"), Lock.c_str());
92
93 return true;
94}
95 /*}}}*/
96// Acquire::~pkgAcquire - Destructor /*{{{*/
97// ---------------------------------------------------------------------
98/* Free our memory, clean up the queues (destroy the workers) */
99pkgAcquire::~pkgAcquire()
100{
101 Shutdown();
102
103 if (LockFD != -1)
104 close(LockFD);
105
106 while (Configs != 0)
107 {
108 MethodConfig *Jnk = Configs;
109 Configs = Configs->Next;
110 delete Jnk;
111 }
112}
113 /*}}}*/
114// Acquire::Shutdown - Clean out the acquire object /*{{{*/
115// ---------------------------------------------------------------------
116/* */
117void pkgAcquire::Shutdown()
118{
119 while (Items.empty() == false)
120 {
121 if (Items[0]->Status == Item::StatFetching)
122 Items[0]->Status = Item::StatError;
123 delete Items[0];
124 }
125
126 while (Queues != 0)
127 {
128 Queue *Jnk = Queues;
129 Queues = Queues->Next;
130 delete Jnk;
131 }
132}
133 /*}}}*/
134// Acquire::Add - Add a new item /*{{{*/
135// ---------------------------------------------------------------------
136/* This puts an item on the acquire list. This list is mainly for tracking
137 item status */
138void pkgAcquire::Add(Item *Itm)
139{
140 Items.push_back(Itm);
141}
142 /*}}}*/
143// Acquire::Remove - Remove a item /*{{{*/
144// ---------------------------------------------------------------------
145/* Remove an item from the acquire list. This is usually not used.. */
146void pkgAcquire::Remove(Item *Itm)
147{
148 Dequeue(Itm);
149
150 for (ItemIterator I = Items.begin(); I != Items.end();)
151 {
152 if (*I == Itm)
153 {
154 Items.erase(I);
155 I = Items.begin();
156 }
157 else
158 ++I;
159 }
160}
161 /*}}}*/
162// Acquire::Add - Add a worker /*{{{*/
163// ---------------------------------------------------------------------
164/* A list of workers is kept so that the select loop can direct their FD
165 usage. */
166void pkgAcquire::Add(Worker *Work)
167{
168 Work->NextAcquire = Workers;
169 Workers = Work;
170}
171 /*}}}*/
172// Acquire::Remove - Remove a worker /*{{{*/
173// ---------------------------------------------------------------------
174/* A worker has died. This can not be done while the select loop is running
175 as it would require that RunFds could handling a changing list state and
176 it cant.. */
177void pkgAcquire::Remove(Worker *Work)
178{
179 if (Running == true)
180 abort();
181
182 Worker **I = &Workers;
183 for (; *I != 0;)
184 {
185 if (*I == Work)
186 *I = (*I)->NextAcquire;
187 else
188 I = &(*I)->NextAcquire;
189 }
190}
191 /*}}}*/
192// Acquire::Enqueue - Queue an URI for fetching /*{{{*/
193// ---------------------------------------------------------------------
194/* This is the entry point for an item. An item calls this function when
195 it is constructed which creates a queue (based on the current queue
196 mode) and puts the item in that queue. If the system is running then
197 the queue might be started. */
198void pkgAcquire::Enqueue(ItemDesc &Item)
199{
200 // Determine which queue to put the item in
201 const MethodConfig *Config;
202 string Name = QueueName(Item.URI,Config);
203 if (Name.empty() == true)
204 return;
205
206 // Find the queue structure
207 Queue *I = Queues;
208 for (; I != 0 && I->Name != Name; I = I->Next);
209 if (I == 0)
210 {
211 I = new Queue(Name,this);
212 I->Next = Queues;
213 Queues = I;
214
215 if (Running == true)
216 I->Startup();
217 }
218
219 // See if this is a local only URI
220 if (Config->LocalOnly == true && Item.Owner->Complete == false)
221 Item.Owner->Local = true;
222 Item.Owner->Status = Item::StatIdle;
223
224 // Queue it into the named queue
225 if(I->Enqueue(Item))
226 ToFetch++;
227
228 // Some trace stuff
229 if (Debug == true)
230 {
231 clog << "Fetching " << Item.URI << endl;
232 clog << " to " << Item.Owner->DestFile << endl;
233 clog << " Queue is: " << Name << endl;
234 }
235}
236 /*}}}*/
237// Acquire::Dequeue - Remove an item from all queues /*{{{*/
238// ---------------------------------------------------------------------
239/* This is called when an item is finished being fetched. It removes it
240 from all the queues */
241void pkgAcquire::Dequeue(Item *Itm)
242{
243 Queue *I = Queues;
244 bool Res = false;
245 for (; I != 0; I = I->Next)
246 Res |= I->Dequeue(Itm);
247
248 if (Debug == true)
249 clog << "Dequeuing " << Itm->DestFile << endl;
250 if (Res == true)
251 ToFetch--;
252}
253 /*}}}*/
254// Acquire::QueueName - Return the name of the queue for this URI /*{{{*/
255// ---------------------------------------------------------------------
256/* The string returned depends on the configuration settings and the
257 method parameters. Given something like http://foo.org/bar it can
258 return http://foo.org or http */
259string pkgAcquire::QueueName(string Uri,MethodConfig const *&Config)
260{
261 URI U(Uri);
262
263 Config = GetConfig(U.Access);
264 if (Config == 0)
265 return string();
266
267 /* Single-Instance methods get exactly one queue per URI. This is
268 also used for the Access queue method */
269 if (Config->SingleInstance == true || QueueMode == QueueAccess)
270 return U.Access;
271
272 return U.Access + ':' + U.Host;
273}
274 /*}}}*/
275// Acquire::GetConfig - Fetch the configuration information /*{{{*/
276// ---------------------------------------------------------------------
277/* This locates the configuration structure for an access method. If
278 a config structure cannot be found a Worker will be created to
279 retrieve it */
280pkgAcquire::MethodConfig *pkgAcquire::GetConfig(string Access)
281{
282 // Search for an existing config
283 MethodConfig *Conf;
284 for (Conf = Configs; Conf != 0; Conf = Conf->Next)
285 if (Conf->Access == Access)
286 return Conf;
287
288 // Create the new config class
289 Conf = new MethodConfig;
290 Conf->Access = Access;
291 Conf->Next = Configs;
292 Configs = Conf;
293
294 // Create the worker to fetch the configuration
295 Worker Work(Conf);
296 if (Work.Start() == false)
297 return 0;
298
299 /* if a method uses DownloadLimit, we switch to SingleInstance mode */
300 if(_config->FindI("Acquire::"+Access+"::Dl-Limit",0) > 0)
301 Conf->SingleInstance = true;
302
303 return Conf;
304}
305 /*}}}*/
306// Acquire::SetFds - Deal with readable FDs /*{{{*/
307// ---------------------------------------------------------------------
308/* Collect FDs that have activity monitors into the fd sets */
309void pkgAcquire::SetFds(int &Fd,fd_set *RSet,fd_set *WSet)
310{
311 for (Worker *I = Workers; I != 0; I = I->NextAcquire)
312 {
313 if (I->InReady == true && I->InFd >= 0)
314 {
315 if (Fd < I->InFd)
316 Fd = I->InFd;
317 FD_SET(I->InFd,RSet);
318 }
319 if (I->OutReady == true && I->OutFd >= 0)
320 {
321 if (Fd < I->OutFd)
322 Fd = I->OutFd;
323 FD_SET(I->OutFd,WSet);
324 }
325 }
326}
327 /*}}}*/
328// Acquire::RunFds - Deal with active FDs /*{{{*/
329// ---------------------------------------------------------------------
330/* Dispatch active FDs over to the proper workers. It is very important
331 that a worker never be erased while this is running! The queue class
332 should never erase a worker except during shutdown processing. */
333void pkgAcquire::RunFds(fd_set *RSet,fd_set *WSet)
334{
335 for (Worker *I = Workers; I != 0; I = I->NextAcquire)
336 {
337 if (I->InFd >= 0 && FD_ISSET(I->InFd,RSet) != 0)
338 I->InFdReady();
339 if (I->OutFd >= 0 && FD_ISSET(I->OutFd,WSet) != 0)
340 I->OutFdReady();
341 }
342}
343 /*}}}*/
344// Acquire::Run - Run the fetch sequence /*{{{*/
345// ---------------------------------------------------------------------
346/* This runs the queues. It manages a select loop for all of the
347 Worker tasks. The workers interact with the queues and items to
348 manage the actual fetch. */
349pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
350{
351 Running = true;
352
353 for (Queue *I = Queues; I != 0; I = I->Next)
354 I->Startup();
355
356 if (Log != 0)
357 Log->Start();
358
359 bool WasCancelled = false;
360
361 // Run till all things have been acquired
362 struct timeval tv;
363 tv.tv_sec = 0;
364 tv.tv_usec = PulseIntervall;
365 while (ToFetch > 0)
366 {
367 fd_set RFds;
368 fd_set WFds;
369 int Highest = 0;
370 FD_ZERO(&RFds);
371 FD_ZERO(&WFds);
372 SetFds(Highest,&RFds,&WFds);
373
374 int Res;
375 do
376 {
377 Res = select(Highest+1,&RFds,&WFds,0,&tv);
378 }
379 while (Res < 0 && errno == EINTR);
380
381 if (Res < 0)
382 {
383 _error->Errno("select","Select has failed");
384 break;
385 }
386
387 RunFds(&RFds,&WFds);
388 if (_error->PendingError() == true)
389 break;
390
391 // Timeout, notify the log class
392 if (Res == 0 || (Log != 0 && Log->Update == true))
393 {
394 tv.tv_usec = PulseIntervall;
395 for (Worker *I = Workers; I != 0; I = I->NextAcquire)
396 I->Pulse();
397 if (Log != 0 && Log->Pulse(this) == false)
398 {
399 WasCancelled = true;
400 break;
401 }
402 }
403 }
404
405 if (Log != 0)
406 Log->Stop();
407
408 // Shut down the acquire bits
409 Running = false;
410 for (Queue *I = Queues; I != 0; I = I->Next)
411 I->Shutdown(false);
412
413 // Shut down the items
414 for (ItemIterator I = Items.begin(); I != Items.end(); ++I)
415 (*I)->Finished();
416
417 if (_error->PendingError())
418 return Failed;
419 if (WasCancelled)
420 return Cancelled;
421 return Continue;
422}
423 /*}}}*/
424// Acquire::Bump - Called when an item is dequeued /*{{{*/
425// ---------------------------------------------------------------------
426/* This routine bumps idle queues in hopes that they will be able to fetch
427 the dequeued item */
428void pkgAcquire::Bump()
429{
430 for (Queue *I = Queues; I != 0; I = I->Next)
431 I->Bump();
432}
433 /*}}}*/
434// Acquire::WorkerStep - Step to the next worker /*{{{*/
435// ---------------------------------------------------------------------
436/* Not inlined to advoid including acquire-worker.h */
437pkgAcquire::Worker *pkgAcquire::WorkerStep(Worker *I)
438{
439 return I->NextAcquire;
440};
441 /*}}}*/
442// Acquire::Clean - Cleans a directory /*{{{*/
443// ---------------------------------------------------------------------
444/* This is a bit simplistic, it looks at every file in the dir and sees
445 if it is part of the download set. */
446bool pkgAcquire::Clean(string Dir)
447{
448 // non-existing directories are by definition clean…
449 if (DirectoryExists(Dir) == false)
450 return true;
451
452 DIR *D = opendir(Dir.c_str());
453 if (D == 0)
454 return _error->Errno("opendir",_("Unable to read %s"),Dir.c_str());
455
456 string StartDir = SafeGetCWD();
457 if (chdir(Dir.c_str()) != 0)
458 {
459 closedir(D);
460 return _error->Errno("chdir",_("Unable to change to %s"),Dir.c_str());
461 }
462
463 for (struct dirent *Dir = readdir(D); Dir != 0; Dir = readdir(D))
464 {
465 // Skip some files..
466 if (strcmp(Dir->d_name,"lock") == 0 ||
467 strcmp(Dir->d_name,"partial") == 0 ||
468 strcmp(Dir->d_name,".") == 0 ||
469 strcmp(Dir->d_name,"..") == 0)
470 continue;
471
472 // Look in the get list
473 ItemCIterator I = Items.begin();
474 for (; I != Items.end(); ++I)
475 if (flNotDir((*I)->DestFile) == Dir->d_name)
476 break;
477
478 // Nothing found, nuke it
479 if (I == Items.end())
480 unlink(Dir->d_name);
481 };
482
483 closedir(D);
484 if (chdir(StartDir.c_str()) != 0)
485 return _error->Errno("chdir",_("Unable to change to %s"),StartDir.c_str());
486 return true;
487}
488 /*}}}*/
489// Acquire::TotalNeeded - Number of bytes to fetch /*{{{*/
490// ---------------------------------------------------------------------
491/* This is the total number of bytes needed */
492unsigned long long pkgAcquire::TotalNeeded()
493{
494 unsigned long long Total = 0;
495 for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
496 Total += (*I)->FileSize;
497 return Total;
498}
499 /*}}}*/
500// Acquire::FetchNeeded - Number of bytes needed to get /*{{{*/
501// ---------------------------------------------------------------------
502/* This is the number of bytes that is not local */
503unsigned long long pkgAcquire::FetchNeeded()
504{
505 unsigned long long Total = 0;
506 for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
507 if ((*I)->Local == false)
508 Total += (*I)->FileSize;
509 return Total;
510}
511 /*}}}*/
512// Acquire::PartialPresent - Number of partial bytes we already have /*{{{*/
513// ---------------------------------------------------------------------
514/* This is the number of bytes that is not local */
515unsigned long long pkgAcquire::PartialPresent()
516{
517 unsigned long long Total = 0;
518 for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
519 if ((*I)->Local == false)
520 Total += (*I)->PartialSize;
521 return Total;
522}
523 /*}}}*/
524// Acquire::UriBegin - Start iterator for the uri list /*{{{*/
525// ---------------------------------------------------------------------
526/* */
527pkgAcquire::UriIterator pkgAcquire::UriBegin()
528{
529 return UriIterator(Queues);
530}
531 /*}}}*/
532// Acquire::UriEnd - End iterator for the uri list /*{{{*/
533// ---------------------------------------------------------------------
534/* */
535pkgAcquire::UriIterator pkgAcquire::UriEnd()
536{
537 return UriIterator(0);
538}
539 /*}}}*/
540// Acquire::MethodConfig::MethodConfig - Constructor /*{{{*/
541// ---------------------------------------------------------------------
542/* */
543pkgAcquire::MethodConfig::MethodConfig()
544{
545 SingleInstance = false;
546 Pipeline = false;
547 SendConfig = false;
548 LocalOnly = false;
549 Removable = false;
550 Next = 0;
551}
552 /*}}}*/
553// Queue::Queue - Constructor /*{{{*/
554// ---------------------------------------------------------------------
555/* */
556pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name),
557 Owner(Owner)
558{
559 Items = 0;
560 Next = 0;
561 Workers = 0;
562 MaxPipeDepth = 1;
563 PipeDepth = 0;
564}
565 /*}}}*/
566// Queue::~Queue - Destructor /*{{{*/
567// ---------------------------------------------------------------------
568/* */
569pkgAcquire::Queue::~Queue()
570{
571 Shutdown(true);
572
573 while (Items != 0)
574 {
575 QItem *Jnk = Items;
576 Items = Items->Next;
577 delete Jnk;
578 }
579}
580 /*}}}*/
581// Queue::Enqueue - Queue an item to the queue /*{{{*/
582// ---------------------------------------------------------------------
583/* */
584bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
585{
586 QItem **I = &Items;
587 // move to the end of the queue and check for duplicates here
588 for (; *I != 0; I = &(*I)->Next)
589 if (Item.URI == (*I)->URI)
590 {
591 Item.Owner->Status = Item::StatDone;
592 return false;
593 }
594
595 // Create a new item
596 QItem *Itm = new QItem;
597 *Itm = Item;
598 Itm->Next = 0;
599 *I = Itm;
600
601 Item.Owner->QueueCounter++;
602 if (Items->Next == 0)
603 Cycle();
604 return true;
605}
606 /*}}}*/
607// Queue::Dequeue - Remove an item from the queue /*{{{*/
608// ---------------------------------------------------------------------
609/* We return true if we hit something */
610bool pkgAcquire::Queue::Dequeue(Item *Owner)
611{
612 if (Owner->Status == pkgAcquire::Item::StatFetching)
613 return _error->Error("Tried to dequeue a fetching object");
614
615 bool Res = false;
616
617 QItem **I = &Items;
618 for (; *I != 0;)
619 {
620 if ((*I)->Owner == Owner)
621 {
622 QItem *Jnk= *I;
623 *I = (*I)->Next;
624 Owner->QueueCounter--;
625 delete Jnk;
626 Res = true;
627 }
628 else
629 I = &(*I)->Next;
630 }
631
632 return Res;
633}
634 /*}}}*/
635// Queue::Startup - Start the worker processes /*{{{*/
636// ---------------------------------------------------------------------
637/* It is possible for this to be called with a pre-existing set of
638 workers. */
639bool pkgAcquire::Queue::Startup()
640{
641 if (Workers == 0)
642 {
643 URI U(Name);
644 pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access);
645 if (Cnf == 0)
646 return false;
647
648 Workers = new Worker(this,Cnf,Owner->Log);
649 Owner->Add(Workers);
650 if (Workers->Start() == false)
651 return false;
652
653 /* When pipelining we commit 10 items. This needs to change when we
654 added other source retry to have cycle maintain a pipeline depth
655 on its own. */
656 if (Cnf->Pipeline == true)
657 MaxPipeDepth = _config->FindI("Acquire::Max-Pipeline-Depth",10);
658 else
659 MaxPipeDepth = 1;
660 }
661
662 return Cycle();
663}
664 /*}}}*/
665// Queue::Shutdown - Shutdown the worker processes /*{{{*/
666// ---------------------------------------------------------------------
667/* If final is true then all workers are eliminated, otherwise only workers
668 that do not need cleanup are removed */
669bool pkgAcquire::Queue::Shutdown(bool Final)
670{
671 // Delete all of the workers
672 pkgAcquire::Worker **Cur = &Workers;
673 while (*Cur != 0)
674 {
675 pkgAcquire::Worker *Jnk = *Cur;
676 if (Final == true || Jnk->GetConf()->NeedsCleanup == false)
677 {
678 *Cur = Jnk->NextQueue;
679 Owner->Remove(Jnk);
680 delete Jnk;
681 }
682 else
683 Cur = &(*Cur)->NextQueue;
684 }
685
686 return true;
687}
688 /*}}}*/
689// Queue::FindItem - Find a URI in the item list /*{{{*/
690// ---------------------------------------------------------------------
691/* */
692pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
693{
694 for (QItem *I = Items; I != 0; I = I->Next)
695 if (I->URI == URI && I->Worker == Owner)
696 return I;
697 return 0;
698}
699 /*}}}*/
700// Queue::ItemDone - Item has been completed /*{{{*/
701// ---------------------------------------------------------------------
702/* The worker signals this which causes the item to be removed from the
703 queue. If this is the last queue instance then it is removed from the
704 main queue too.*/
705bool pkgAcquire::Queue::ItemDone(QItem *Itm)
706{
707 PipeDepth--;
708 if (Itm->Owner->Status == pkgAcquire::Item::StatFetching)
709 Itm->Owner->Status = pkgAcquire::Item::StatDone;
710
711 if (Itm->Owner->QueueCounter <= 1)
712 Owner->Dequeue(Itm->Owner);
713 else
714 {
715 Dequeue(Itm->Owner);
716 Owner->Bump();
717 }
718
719 return Cycle();
720}
721 /*}}}*/
722// Queue::Cycle - Queue new items into the method /*{{{*/
723// ---------------------------------------------------------------------
724/* This locates a new idle item and sends it to the worker. If pipelining
725 is enabled then it keeps the pipe full. */
726bool pkgAcquire::Queue::Cycle()
727{
728 if (Items == 0 || Workers == 0)
729 return true;
730
731 if (PipeDepth < 0)
732 return _error->Error("Pipedepth failure");
733
734 // Look for a queable item
735 QItem *I = Items;
736 while (PipeDepth < (signed)MaxPipeDepth)
737 {
738 for (; I != 0; I = I->Next)
739 if (I->Owner->Status == pkgAcquire::Item::StatIdle)
740 break;
741
742 // Nothing to do, queue is idle.
743 if (I == 0)
744 return true;
745
746 I->Worker = Workers;
747 I->Owner->Status = pkgAcquire::Item::StatFetching;
748 PipeDepth++;
749 if (Workers->QueueItem(I) == false)
750 return false;
751 }
752
753 return true;
754}
755 /*}}}*/
756// Queue::Bump - Fetch any pending objects if we are idle /*{{{*/
757// ---------------------------------------------------------------------
758/* This is called when an item in multiple queues is dequeued */
759void pkgAcquire::Queue::Bump()
760{
761 Cycle();
762}
763 /*}}}*/
764// AcquireStatus::pkgAcquireStatus - Constructor /*{{{*/
765// ---------------------------------------------------------------------
766/* */
767pkgAcquireStatus::pkgAcquireStatus() : Update(true), MorePulses(false)
768{
769 Start();
770}
771 /*}}}*/
772// AcquireStatus::Pulse - Called periodically /*{{{*/
773// ---------------------------------------------------------------------
774/* This computes some internal state variables for the derived classes to
775 use. It generates the current downloaded bytes and total bytes to download
776 as well as the current CPS estimate. */
777bool pkgAcquireStatus::Pulse(pkgAcquire *Owner)
778{
779 TotalBytes = 0;
780 CurrentBytes = 0;
781 TotalItems = 0;
782 CurrentItems = 0;
783
784 // Compute the total number of bytes to fetch
785 unsigned int Unknown = 0;
786 unsigned int Count = 0;
787 for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin(); I != Owner->ItemsEnd();
788 ++I, ++Count)
789 {
790 TotalItems++;
791 if ((*I)->Status == pkgAcquire::Item::StatDone)
792 ++CurrentItems;
793
794 // Totally ignore local items
795 if ((*I)->Local == true)
796 continue;
797
798 TotalBytes += (*I)->FileSize;
799 if ((*I)->Complete == true)
800 CurrentBytes += (*I)->FileSize;
801 if ((*I)->FileSize == 0 && (*I)->Complete == false)
802 ++Unknown;
803 }
804
805 // Compute the current completion
806 unsigned long ResumeSize = 0;
807 for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0;
808 I = Owner->WorkerStep(I))
809 if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false)
810 {
811 CurrentBytes += I->CurrentSize;
812 ResumeSize += I->ResumePoint;
813
814 // Files with unknown size always have 100% completion
815 if (I->CurrentItem->Owner->FileSize == 0 &&
816 I->CurrentItem->Owner->Complete == false)
817 TotalBytes += I->CurrentSize;
818 }
819
820 // Normalize the figures and account for unknown size downloads
821 if (TotalBytes <= 0)
822 TotalBytes = 1;
823 if (Unknown == Count)
824 TotalBytes = Unknown;
825
826 // Wha?! Is not supposed to happen.
827 if (CurrentBytes > TotalBytes)
828 CurrentBytes = TotalBytes;
829
830 // Compute the CPS
831 struct timeval NewTime;
832 gettimeofday(&NewTime,0);
833 if ((NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec) ||
834 NewTime.tv_sec - Time.tv_sec > 6)
835 {
836 double Delta = NewTime.tv_sec - Time.tv_sec +
837 (NewTime.tv_usec - Time.tv_usec)/1000000.0;
838
839 // Compute the CPS value
840 if (Delta < 0.01)
841 CurrentCPS = 0;
842 else
843 CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta;
844 LastBytes = CurrentBytes - ResumeSize;
845 ElapsedTime = (unsigned long)Delta;
846 Time = NewTime;
847 }
848
849 int fd = _config->FindI("APT::Status-Fd",-1);
850 if(fd > 0)
851 {
852 ostringstream status;
853
854 char msg[200];
855 long i = CurrentItems < TotalItems ? CurrentItems + 1 : CurrentItems;
856 unsigned long ETA =
857 (unsigned long)((TotalBytes - CurrentBytes) / CurrentCPS);
858
859 // only show the ETA if it makes sense
860 if (ETA > 0 && ETA < 172800 /* two days */ )
861 snprintf(msg,sizeof(msg), _("Retrieving file %li of %li (%s remaining)"), i, TotalItems, TimeToStr(ETA).c_str());
862 else
863 snprintf(msg,sizeof(msg), _("Retrieving file %li of %li"), i, TotalItems);
864
865
866
867 // build the status str
868 status << "dlstatus:" << i
869 << ":" << (CurrentBytes/float(TotalBytes)*100.0)
870 << ":" << msg
871 << endl;
872 write(fd, status.str().c_str(), status.str().size());
873 }
874
875 return true;
876}
877 /*}}}*/
878// AcquireStatus::Start - Called when the download is started /*{{{*/
879// ---------------------------------------------------------------------
880/* We just reset the counters */
881void pkgAcquireStatus::Start()
882{
883 gettimeofday(&Time,0);
884 gettimeofday(&StartTime,0);
885 LastBytes = 0;
886 CurrentCPS = 0;
887 CurrentBytes = 0;
888 TotalBytes = 0;
889 FetchedBytes = 0;
890 ElapsedTime = 0;
891 TotalItems = 0;
892 CurrentItems = 0;
893}
894 /*}}}*/
895// AcquireStatus::Stop - Finished downloading /*{{{*/
896// ---------------------------------------------------------------------
897/* This accurately computes the elapsed time and the total overall CPS. */
898void pkgAcquireStatus::Stop()
899{
900 // Compute the CPS and elapsed time
901 struct timeval NewTime;
902 gettimeofday(&NewTime,0);
903
904 double Delta = NewTime.tv_sec - StartTime.tv_sec +
905 (NewTime.tv_usec - StartTime.tv_usec)/1000000.0;
906
907 // Compute the CPS value
908 if (Delta < 0.01)
909 CurrentCPS = 0;
910 else
911 CurrentCPS = FetchedBytes/Delta;
912 LastBytes = CurrentBytes;
913 ElapsedTime = (unsigned int)Delta;
914}
915 /*}}}*/
916// AcquireStatus::Fetched - Called when a byte set has been fetched /*{{{*/
917// ---------------------------------------------------------------------
918/* This is used to get accurate final transfer rate reporting. */
919void pkgAcquireStatus::Fetched(unsigned long Size,unsigned long Resume)
920{
921 FetchedBytes += Size - Resume;
922}
923 /*}}}*/