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