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