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