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