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