]> git.saurik.com Git - apt.git/blame - apt-pkg/acquire.cc
Merge branch 'debian/sid' into debian/experimental
[apt.git] / apt-pkg / acquire.cc
CommitLineData
0118833a
AL
1// -*- mode: cpp; mode: fold -*-
2// Description /*{{{*/
1b480911 3// $Id: acquire.cc,v 1.50 2004/03/17 05:17:11 mdz Exp $
0118833a
AL
4/* ######################################################################
5
6 Acquire - File Acquiration
7
1e3f4083 8 The core element for the schedule system is the concept of a named
0a8a80e5 9 queue. Each queue is unique and each queue has a name derived from the
1e3f4083 10 URI. The degree of paralization can be controlled by how the queue
0a8a80e5
AL
11 name is derived from the URI.
12
0118833a
AL
13 ##################################################################### */
14 /*}}}*/
15// Include Files /*{{{*/
ea542140
DK
16#include <config.h>
17
0118833a
AL
18#include <apt-pkg/acquire.h>
19#include <apt-pkg/acquire-item.h>
20#include <apt-pkg/acquire-worker.h>
0a8a80e5
AL
21#include <apt-pkg/configuration.h>
22#include <apt-pkg/error.h>
cdcc6d34 23#include <apt-pkg/strutl.h>
1cd1c398 24#include <apt-pkg/fileutl.h>
8267fe24 25
453b82a3
DK
26#include <string>
27#include <vector>
b4fc9b6f 28#include <iostream>
75ef8f14 29#include <sstream>
04a54261
DK
30#include <iomanip>
31
526334a0 32#include <stdio.h>
453b82a3
DK
33#include <stdlib.h>
34#include <string.h>
35#include <unistd.h>
04a54261
DK
36#include <pwd.h>
37#include <grp.h>
7a7fa5f0 38#include <dirent.h>
8267fe24 39#include <sys/time.h>
453b82a3 40#include <sys/select.h>
524f8105 41#include <errno.h>
56472095 42#include <sys/stat.h>
04a54261 43#include <sys/types.h>
ea542140
DK
44
45#include <apti18n.h>
0118833a
AL
46 /*}}}*/
47
b4fc9b6f
AL
48using namespace std;
49
0118833a
AL
50// Acquire::pkgAcquire - Constructor /*{{{*/
51// ---------------------------------------------------------------------
93bf083d 52/* We grab some runtime state from the configuration space */
5efbd596 53pkgAcquire::pkgAcquire() : LockFD(-1), Queues(0), Workers(0), Configs(0), Log(NULL), ToFetch(0),
1cd1c398 54 Debug(_config->FindB("Debug::pkgAcquire",false)),
5efbd596 55 Running(false)
0118833a 56{
1cd1c398 57 string const Mode = _config->Find("Acquire::Queue-Mode","host");
0a8a80e5
AL
58 if (strcasecmp(Mode.c_str(),"host") == 0)
59 QueueMode = QueueHost;
60 if (strcasecmp(Mode.c_str(),"access") == 0)
1cd1c398
DK
61 QueueMode = QueueAccess;
62}
04a54261
DK
63pkgAcquire::pkgAcquire(pkgAcquireStatus *Progress) : LockFD(-1), Queues(0), Workers(0),
64 Configs(0), Log(NULL), ToFetch(0),
1cd1c398 65 Debug(_config->FindB("Debug::pkgAcquire",false)),
5efbd596 66 Running(false)
1cd1c398
DK
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;
04a54261 73 SetLog(Progress);
1cd1c398
DK
74}
75 /*}}}*/
04a54261
DK
76// Acquire::GetLock - lock directory and prepare for action /*{{{*/
77static bool SetupAPTPartialDirectory(std::string const &grand, std::string const &parent)
1cd1c398 78{
04a54261
DK
79 std::string const partial = parent + "partial";
80 if (CreateAPTDirectoryIfNeeded(grand, partial) == false &&
81 CreateAPTDirectoryIfNeeded(parent, partial) == false)
82 return false;
0a8a80e5 83
04a54261
DK
84 if (getuid() == 0) // if we aren't root, we can't chown, so don't try it
85 {
9983999d
MV
86 std::string SandboxUser = _config->Find("APT::Sandbox::User");
87 struct passwd *pw = getpwnam(SandboxUser.c_str());
04a54261
DK
88 struct group *gr = getgrnam("root");
89 if (pw != NULL && gr != NULL && chown(partial.c_str(), pw->pw_uid, gr->gr_gid) != 0)
9983999d 90 _error->WarningE("SetupAPTPartialDirectory", "chown to %s:root of directory %s failed", SandboxUser.c_str(), partial.c_str());
04a54261
DK
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}
97bool pkgAcquire::Setup(pkgAcquireStatus *Progress, string const &Lock)
98{
99 Log = Progress;
100 if (Lock.empty())
43acd019
DK
101 {
102 string const listDir = _config->FindDir("Dir::State::lists");
04a54261
DK
103 if (SetupAPTPartialDirectory(_config->FindDir("Dir::State"), listDir) == false)
104 return _error->Errno("Acquire", _("List directory %spartial is missing."), listDir.c_str());
43acd019 105 string const archivesDir = _config->FindDir("Dir::Cache::Archives");
04a54261
DK
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}
112bool pkgAcquire::GetLock(std::string const &Lock)
113{
114 if (Lock.empty() == true)
115 return false;
9c2c9c24 116
04a54261
DK
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");
9c2c9c24 120
04a54261
DK
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)
43acd019
DK
129 return _error->Errno("Acquire", _("Archives directory %spartial is missing."), archivesDir.c_str());
130 }
1cd1c398 131
04a54261 132 if (_config->FindB("Debug::NoLocking", false) == true)
1cd1c398
DK
133 return true;
134
135 // Lock the directory this acquire object will work in
04a54261 136 LockFD = ::GetLock(flCombine(Lock, "lock"));
1cd1c398
DK
137 if (LockFD == -1)
138 return _error->Error(_("Unable to lock directory %s"), Lock.c_str());
139
140 return true;
141}
142 /*}}}*/
0118833a
AL
143// Acquire::~pkgAcquire - Destructor /*{{{*/
144// ---------------------------------------------------------------------
93bf083d 145/* Free our memory, clean up the queues (destroy the workers) */
0118833a
AL
146pkgAcquire::~pkgAcquire()
147{
459681d3 148 Shutdown();
1cd1c398
DK
149
150 if (LockFD != -1)
151 close(LockFD);
152
3b5421b4
AL
153 while (Configs != 0)
154 {
155 MethodConfig *Jnk = Configs;
156 Configs = Configs->Next;
157 delete Jnk;
158 }
281daf46
AL
159}
160 /*}}}*/
8e5fc8f5 161// Acquire::Shutdown - Clean out the acquire object /*{{{*/
281daf46
AL
162// ---------------------------------------------------------------------
163/* */
164void pkgAcquire::Shutdown()
165{
f7f0d6c7 166 while (Items.empty() == false)
1b480911
AL
167 {
168 if (Items[0]->Status == Item::StatFetching)
169 Items[0]->Status = Item::StatError;
281daf46 170 delete Items[0];
1b480911 171 }
0a8a80e5
AL
172
173 while (Queues != 0)
174 {
175 Queue *Jnk = Queues;
176 Queues = Queues->Next;
177 delete Jnk;
178 }
0118833a
AL
179}
180 /*}}}*/
181// Acquire::Add - Add a new item /*{{{*/
182// ---------------------------------------------------------------------
93bf083d
AL
183/* This puts an item on the acquire list. This list is mainly for tracking
184 item status */
0118833a
AL
185void pkgAcquire::Add(Item *Itm)
186{
187 Items.push_back(Itm);
188}
189 /*}}}*/
190// Acquire::Remove - Remove a item /*{{{*/
191// ---------------------------------------------------------------------
93bf083d 192/* Remove an item from the acquire list. This is usually not used.. */
0118833a
AL
193void pkgAcquire::Remove(Item *Itm)
194{
a3eaf954
AL
195 Dequeue(Itm);
196
753b3525 197 for (ItemIterator I = Items.begin(); I != Items.end();)
0118833a
AL
198 {
199 if (*I == Itm)
b4fc9b6f 200 {
0118833a 201 Items.erase(I);
b4fc9b6f
AL
202 I = Items.begin();
203 }
753b3525 204 else
f7f0d6c7 205 ++I;
8267fe24 206 }
0118833a
AL
207}
208 /*}}}*/
0a8a80e5
AL
209// Acquire::Add - Add a worker /*{{{*/
210// ---------------------------------------------------------------------
93bf083d
AL
211/* A list of workers is kept so that the select loop can direct their FD
212 usage. */
0a8a80e5
AL
213void pkgAcquire::Add(Worker *Work)
214{
215 Work->NextAcquire = Workers;
216 Workers = Work;
217}
218 /*}}}*/
219// Acquire::Remove - Remove a worker /*{{{*/
220// ---------------------------------------------------------------------
93bf083d
AL
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
1e3f4083 223 it can't.. */
0a8a80e5
AL
224void pkgAcquire::Remove(Worker *Work)
225{
93bf083d
AL
226 if (Running == true)
227 abort();
228
0a8a80e5
AL
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 /*}}}*/
0118833a
AL
239// Acquire::Enqueue - Queue an URI for fetching /*{{{*/
240// ---------------------------------------------------------------------
93bf083d 241/* This is the entry point for an item. An item calls this function when
281daf46 242 it is constructed which creates a queue (based on the current queue
93bf083d
AL
243 mode) and puts the item in that queue. If the system is running then
244 the queue might be started. */
8267fe24 245void pkgAcquire::Enqueue(ItemDesc &Item)
0118833a 246{
0a8a80e5 247 // Determine which queue to put the item in
e331f6ed
AL
248 const MethodConfig *Config;
249 string Name = QueueName(Item.URI,Config);
0a8a80e5
AL
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;
93bf083d
AL
261
262 if (Running == true)
263 I->Startup();
0a8a80e5 264 }
bfd22fc0 265
e331f6ed
AL
266 // See if this is a local only URI
267 if (Config->LocalOnly == true && Item.Owner->Complete == false)
268 Item.Owner->Local = true;
8267fe24 269 Item.Owner->Status = Item::StatIdle;
0a8a80e5
AL
270
271 // Queue it into the named queue
c03462c6
MV
272 if(I->Enqueue(Item))
273 ToFetch++;
274
0a8a80e5
AL
275 // Some trace stuff
276 if (Debug == true)
277 {
8267fe24
AL
278 clog << "Fetching " << Item.URI << endl;
279 clog << " to " << Item.Owner->DestFile << endl;
e331f6ed 280 clog << " Queue is: " << Name << endl;
0a8a80e5 281 }
3b5421b4
AL
282}
283 /*}}}*/
0a8a80e5 284// Acquire::Dequeue - Remove an item from all queues /*{{{*/
3b5421b4 285// ---------------------------------------------------------------------
93bf083d
AL
286/* This is called when an item is finished being fetched. It removes it
287 from all the queues */
0a8a80e5
AL
288void pkgAcquire::Dequeue(Item *Itm)
289{
290 Queue *I = Queues;
bfd22fc0 291 bool Res = false;
93bf083d
AL
292 if (Debug == true)
293 clog << "Dequeuing " << Itm->DestFile << endl;
5674f6b3
RG
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
bfd22fc0
AL
305 if (Res == true)
306 ToFetch--;
0a8a80e5
AL
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 */
e331f6ed 314string pkgAcquire::QueueName(string Uri,MethodConfig const *&Config)
3b5421b4 315{
93bf083d
AL
316 URI U(Uri);
317
e331f6ed 318 Config = GetConfig(U.Access);
0a8a80e5
AL
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)
5674f6b3
RG
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;
93bf083d 347
5674f6b3 348 return FullQueueName;
0118833a
AL
349}
350 /*}}}*/
3b5421b4
AL
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 */
0a8a80e5 356pkgAcquire::MethodConfig *pkgAcquire::GetConfig(string Access)
3b5421b4
AL
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;
0118833a 369
3b5421b4
AL
370 // Create the worker to fetch the configuration
371 Worker Work(Conf);
372 if (Work.Start() == false)
373 return 0;
7c6e2dc7
MV
374
375 /* if a method uses DownloadLimit, we switch to SingleInstance mode */
4b65cc13 376 if(_config->FindI("Acquire::"+Access+"::Dl-Limit",0) > 0)
7c6e2dc7
MV
377 Conf->SingleInstance = true;
378
3b5421b4
AL
379 return Conf;
380}
381 /*}}}*/
0a8a80e5
AL
382// Acquire::SetFds - Deal with readable FDs /*{{{*/
383// ---------------------------------------------------------------------
384/* Collect FDs that have activity monitors into the fd sets */
385void 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// ---------------------------------------------------------------------
93bf083d
AL
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. */
0a8a80e5
AL
409void 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. */
1c5f7e5f 425pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
0a8a80e5 426{
8b89e57f
AL
427 Running = true;
428
0a8a80e5
AL
429 for (Queue *I = Queues; I != 0; I = I->Next)
430 I->Startup();
431
b98f2859
AL
432 if (Log != 0)
433 Log->Start();
434
024d1123
AL
435 bool WasCancelled = false;
436
0a8a80e5 437 // Run till all things have been acquired
8267fe24
AL
438 struct timeval tv;
439 tv.tv_sec = 0;
1c5f7e5f 440 tv.tv_usec = PulseIntervall;
0a8a80e5
AL
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
b0db36b1
AL
450 int Res;
451 do
452 {
453 Res = select(Highest+1,&RFds,&WFds,0,&tv);
454 }
455 while (Res < 0 && errno == EINTR);
456
8267fe24 457 if (Res < 0)
8b89e57f 458 {
8267fe24
AL
459 _error->Errno("select","Select has failed");
460 break;
8b89e57f 461 }
93bf083d 462
0a8a80e5 463 RunFds(&RFds,&WFds);
93bf083d
AL
464 if (_error->PendingError() == true)
465 break;
8267fe24
AL
466
467 // Timeout, notify the log class
468 if (Res == 0 || (Log != 0 && Log->Update == true))
469 {
1c5f7e5f 470 tv.tv_usec = PulseIntervall;
8267fe24
AL
471 for (Worker *I = Workers; I != 0; I = I->NextAcquire)
472 I->Pulse();
024d1123
AL
473 if (Log != 0 && Log->Pulse(this) == false)
474 {
475 WasCancelled = true;
476 break;
477 }
8267fe24 478 }
0a8a80e5 479 }
be4401bf 480
b98f2859
AL
481 if (Log != 0)
482 Log->Stop();
483
be4401bf
AL
484 // Shut down the acquire bits
485 Running = false;
0a8a80e5 486 for (Queue *I = Queues; I != 0; I = I->Next)
8e5fc8f5 487 I->Shutdown(false);
0a8a80e5 488
ab559b35 489 // Shut down the items
f7f0d6c7 490 for (ItemIterator I = Items.begin(); I != Items.end(); ++I)
8e5fc8f5 491 (*I)->Finished();
ab559b35 492
024d1123
AL
493 if (_error->PendingError())
494 return Failed;
495 if (WasCancelled)
496 return Cancelled;
497 return Continue;
93bf083d
AL
498}
499 /*}}}*/
be4401bf 500// Acquire::Bump - Called when an item is dequeued /*{{{*/
93bf083d
AL
501// ---------------------------------------------------------------------
502/* This routine bumps idle queues in hopes that they will be able to fetch
503 the dequeued item */
504void pkgAcquire::Bump()
505{
be4401bf
AL
506 for (Queue *I = Queues; I != 0; I = I->Next)
507 I->Bump();
0a8a80e5
AL
508}
509 /*}}}*/
8267fe24
AL
510// Acquire::WorkerStep - Step to the next worker /*{{{*/
511// ---------------------------------------------------------------------
512/* Not inlined to advoid including acquire-worker.h */
513pkgAcquire::Worker *pkgAcquire::WorkerStep(Worker *I)
514{
515 return I->NextAcquire;
d3e8fbb3 516}
8267fe24 517 /*}}}*/
a6568219 518// Acquire::Clean - Cleans a directory /*{{{*/
7a7fa5f0
AL
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. */
522bool pkgAcquire::Clean(string Dir)
523{
95b5f6c1
DK
524 // non-existing directories are by definition clean…
525 if (DirectoryExists(Dir) == false)
526 return true;
527
10ecfe4f
MV
528 if(Dir == "/")
529 return _error->Error(_("Clean of %s is not supported"), Dir.c_str());
530
7a7fa5f0
AL
531 DIR *D = opendir(Dir.c_str());
532 if (D == 0)
b2e465d6 533 return _error->Errno("opendir",_("Unable to read %s"),Dir.c_str());
7a7fa5f0
AL
534
535 string StartDir = SafeGetCWD();
536 if (chdir(Dir.c_str()) != 0)
537 {
538 closedir(D);
b2e465d6 539 return _error->Errno("chdir",_("Unable to change to %s"),Dir.c_str());
7a7fa5f0
AL
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
b4fc9b6f 552 ItemCIterator I = Items.begin();
f7f0d6c7 553 for (; I != Items.end(); ++I)
7a7fa5f0
AL
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
7a7fa5f0 562 closedir(D);
3c8cda8b
MV
563 if (chdir(StartDir.c_str()) != 0)
564 return _error->Errno("chdir",_("Unable to change to %s"),StartDir.c_str());
7a7fa5f0
AL
565 return true;
566}
567 /*}}}*/
a6568219
AL
568// Acquire::TotalNeeded - Number of bytes to fetch /*{{{*/
569// ---------------------------------------------------------------------
570/* This is the total number of bytes needed */
a02db58f 571APT_PURE unsigned long long pkgAcquire::TotalNeeded()
a6568219 572{
a3c4c81a 573 unsigned long long Total = 0;
f7f0d6c7 574 for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
a6568219
AL
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 */
a02db58f 582APT_PURE unsigned long long pkgAcquire::FetchNeeded()
a6568219 583{
a3c4c81a 584 unsigned long long Total = 0;
f7f0d6c7 585 for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
a6568219
AL
586 if ((*I)->Local == false)
587 Total += (*I)->FileSize;
588 return Total;
589}
590 /*}}}*/
6b1ff003
AL
591// Acquire::PartialPresent - Number of partial bytes we already have /*{{{*/
592// ---------------------------------------------------------------------
593/* This is the number of bytes that is not local */
a02db58f 594APT_PURE unsigned long long pkgAcquire::PartialPresent()
6b1ff003 595{
a3c4c81a 596 unsigned long long Total = 0;
f7f0d6c7 597 for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
6b1ff003
AL
598 if ((*I)->Local == false)
599 Total += (*I)->PartialSize;
600 return Total;
601}
92fcbfc1 602 /*}}}*/
8e5fc8f5 603// Acquire::UriBegin - Start iterator for the uri list /*{{{*/
f7a08e33
AL
604// ---------------------------------------------------------------------
605/* */
606pkgAcquire::UriIterator pkgAcquire::UriBegin()
607{
608 return UriIterator(Queues);
609}
610 /*}}}*/
8e5fc8f5 611// Acquire::UriEnd - End iterator for the uri list /*{{{*/
f7a08e33
AL
612// ---------------------------------------------------------------------
613/* */
614pkgAcquire::UriIterator pkgAcquire::UriEnd()
615{
616 return UriIterator(0);
617}
618 /*}}}*/
e331f6ed
AL
619// Acquire::MethodConfig::MethodConfig - Constructor /*{{{*/
620// ---------------------------------------------------------------------
621/* */
25613a61
DK
622pkgAcquire::MethodConfig::MethodConfig() : d(NULL), Next(0), SingleInstance(false),
623 Pipeline(false), SendConfig(false), LocalOnly(false), NeedsCleanup(false),
624 Removable(false)
e331f6ed 625{
e331f6ed
AL
626}
627 /*}}}*/
0a8a80e5
AL
628// Queue::Queue - Constructor /*{{{*/
629// ---------------------------------------------------------------------
630/* */
25613a61
DK
631pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : d(NULL), Next(0),
632 Name(Name), Items(0), Workers(0), Owner(Owner), PipeDepth(0), MaxPipeDepth(1)
0a8a80e5 633{
0a8a80e5
AL
634}
635 /*}}}*/
636// Queue::~Queue - Destructor /*{{{*/
637// ---------------------------------------------------------------------
638/* */
639pkgAcquire::Queue::~Queue()
640{
8e5fc8f5 641 Shutdown(true);
0a8a80e5
AL
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/* */
c03462c6 654bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
0a8a80e5 655{
7a1b1f8b 656 QItem **I = &Items;
c03462c6
MV
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
0a8a80e5 665 // Create a new item
7a1b1f8b
AL
666 QItem *Itm = new QItem;
667 *Itm = Item;
668 Itm->Next = 0;
669 *I = Itm;
0a8a80e5 670
8267fe24 671 Item.Owner->QueueCounter++;
93bf083d
AL
672 if (Items->Next == 0)
673 Cycle();
c03462c6 674 return true;
0a8a80e5
AL
675}
676 /*}}}*/
c88edf1d 677// Queue::Dequeue - Remove an item from the queue /*{{{*/
0a8a80e5 678// ---------------------------------------------------------------------
b185acc2 679/* We return true if we hit something */
bfd22fc0 680bool pkgAcquire::Queue::Dequeue(Item *Owner)
0a8a80e5 681{
b185acc2
AL
682 if (Owner->Status == pkgAcquire::Item::StatFetching)
683 return _error->Error("Tried to dequeue a fetching object");
684
bfd22fc0
AL
685 bool Res = false;
686
0a8a80e5
AL
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;
bfd22fc0 696 Res = true;
0a8a80e5
AL
697 }
698 else
699 I = &(*I)->Next;
700 }
bfd22fc0
AL
701
702 return Res;
0a8a80e5
AL
703}
704 /*}}}*/
705// Queue::Startup - Start the worker processes /*{{{*/
706// ---------------------------------------------------------------------
8e5fc8f5
AL
707/* It is possible for this to be called with a pre-existing set of
708 workers. */
0a8a80e5
AL
709bool pkgAcquire::Queue::Startup()
710{
8e5fc8f5
AL
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)
6ce72612 727 MaxPipeDepth = _config->FindI("Acquire::Max-Pipeline-Depth",10);
8e5fc8f5
AL
728 else
729 MaxPipeDepth = 1;
730 }
5cb5d8dc 731
93bf083d 732 return Cycle();
0a8a80e5
AL
733}
734 /*}}}*/
735// Queue::Shutdown - Shutdown the worker processes /*{{{*/
736// ---------------------------------------------------------------------
8e5fc8f5
AL
737/* If final is true then all workers are eliminated, otherwise only workers
738 that do not need cleanup are removed */
739bool pkgAcquire::Queue::Shutdown(bool Final)
0a8a80e5
AL
740{
741 // Delete all of the workers
8e5fc8f5
AL
742 pkgAcquire::Worker **Cur = &Workers;
743 while (*Cur != 0)
0a8a80e5 744 {
8e5fc8f5
AL
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;
0a8a80e5
AL
754 }
755
756 return true;
3b5421b4
AL
757}
758 /*}}}*/
7d8afa39 759// Queue::FindItem - Find a URI in the item list /*{{{*/
c88edf1d
AL
760// ---------------------------------------------------------------------
761/* */
762pkgAcquire::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
93bf083d
AL
773 queue. If this is the last queue instance then it is removed from the
774 main queue too.*/
c88edf1d
AL
775bool pkgAcquire::Queue::ItemDone(QItem *Itm)
776{
b185acc2 777 PipeDepth--;
db890fdb
AL
778 if (Itm->Owner->Status == pkgAcquire::Item::StatFetching)
779 Itm->Owner->Status = pkgAcquire::Item::StatDone;
780
93bf083d
AL
781 if (Itm->Owner->QueueCounter <= 1)
782 Owner->Dequeue(Itm->Owner);
783 else
784 {
785 Dequeue(Itm->Owner);
786 Owner->Bump();
787 }
c88edf1d 788
93bf083d
AL
789 return Cycle();
790}
791 /*}}}*/
792// Queue::Cycle - Queue new items into the method /*{{{*/
793// ---------------------------------------------------------------------
b185acc2
AL
794/* This locates a new idle item and sends it to the worker. If pipelining
795 is enabled then it keeps the pipe full. */
93bf083d
AL
796bool pkgAcquire::Queue::Cycle()
797{
798 if (Items == 0 || Workers == 0)
c88edf1d
AL
799 return true;
800
e7432370
AL
801 if (PipeDepth < 0)
802 return _error->Error("Pipedepth failure");
803
93bf083d
AL
804 // Look for a queable item
805 QItem *I = Items;
e7432370 806 while (PipeDepth < (signed)MaxPipeDepth)
b185acc2
AL
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;
e7432370 818 PipeDepth++;
b185acc2
AL
819 if (Workers->QueueItem(I) == false)
820 return false;
821 }
93bf083d 822
b185acc2 823 return true;
c88edf1d
AL
824}
825 /*}}}*/
be4401bf
AL
826// Queue::Bump - Fetch any pending objects if we are idle /*{{{*/
827// ---------------------------------------------------------------------
b185acc2 828/* This is called when an item in multiple queues is dequeued */
be4401bf
AL
829void pkgAcquire::Queue::Bump()
830{
b185acc2 831 Cycle();
be4401bf
AL
832}
833 /*}}}*/
b98f2859
AL
834// AcquireStatus::pkgAcquireStatus - Constructor /*{{{*/
835// ---------------------------------------------------------------------
836/* */
25613a61 837pkgAcquireStatus::pkgAcquireStatus() : d(NULL), Percent(0), Update(true), MorePulses(false)
b98f2859
AL
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. */
024d1123 847bool pkgAcquireStatus::Pulse(pkgAcquire *Owner)
b98f2859
AL
848{
849 TotalBytes = 0;
850 CurrentBytes = 0;
d568ed2d
AL
851 TotalItems = 0;
852 CurrentItems = 0;
b98f2859
AL
853
854 // Compute the total number of bytes to fetch
855 unsigned int Unknown = 0;
856 unsigned int Count = 0;
c6e9cc58
MV
857 bool UnfetchedReleaseFiles = false;
858 for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin();
859 I != Owner->ItemsEnd();
f7f0d6c7 860 ++I, ++Count)
b98f2859 861 {
d568ed2d
AL
862 TotalItems++;
863 if ((*I)->Status == pkgAcquire::Item::StatDone)
f7f0d6c7 864 ++CurrentItems;
d568ed2d 865
a6568219
AL
866 // Totally ignore local items
867 if ((*I)->Local == true)
868 continue;
b2e465d6 869
d0cfa8ad
MV
870 // see if the method tells us to expect more
871 TotalItems += (*I)->ExpectedAdditionalItems;
872
c6e9cc58
MV
873 // check if there are unfetched Release files
874 if ((*I)->Complete == false && (*I)->ExpectedAdditionalItems > 0)
875 UnfetchedReleaseFiles = true;
876
b98f2859
AL
877 TotalBytes += (*I)->FileSize;
878 if ((*I)->Complete == true)
879 CurrentBytes += (*I)->FileSize;
880 if ((*I)->FileSize == 0 && (*I)->Complete == false)
f7f0d6c7 881 ++Unknown;
b98f2859
AL
882 }
883
884 // Compute the current completion
dbbc5494 885 unsigned long long ResumeSize = 0;
b98f2859
AL
886 for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0;
887 I = Owner->WorkerStep(I))
c62f7898 888 {
b98f2859 889 if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false)
aa0e1101
AL
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 }
c62f7898 899 }
aa0e1101 900
b98f2859
AL
901 // Normalize the figures and account for unknown size downloads
902 if (TotalBytes <= 0)
903 TotalBytes = 1;
904 if (Unknown == Count)
905 TotalBytes = Unknown;
18ef0a78
AL
906
907 // Wha?! Is not supposed to happen.
908 if (CurrentBytes > TotalBytes)
909 CurrentBytes = TotalBytes;
96c6cab1
MV
910
911 // debug
912 if (_config->FindB("Debug::acquire::progress", false) == true)
913 std::clog << " Bytes: "
914 << SizeToStr(CurrentBytes) << " / " << SizeToStr(TotalBytes)
915 << std::endl;
b98f2859
AL
916
917 // Compute the CPS
918 struct timeval NewTime;
919 gettimeofday(&NewTime,0);
2ec1674d 920 if ((NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec) ||
b98f2859
AL
921 NewTime.tv_sec - Time.tv_sec > 6)
922 {
f17ac097
AL
923 double Delta = NewTime.tv_sec - Time.tv_sec +
924 (NewTime.tv_usec - Time.tv_usec)/1000000.0;
b98f2859 925
b98f2859 926 // Compute the CPS value
f17ac097 927 if (Delta < 0.01)
e331f6ed
AL
928 CurrentCPS = 0;
929 else
aa0e1101
AL
930 CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta;
931 LastBytes = CurrentBytes - ResumeSize;
dbbc5494 932 ElapsedTime = (unsigned long long)Delta;
b98f2859
AL
933 Time = NewTime;
934 }
024d1123 935
c6e9cc58
MV
936 // calculate the percentage, if we have too little data assume 1%
937 if (TotalBytes > 0 && UnfetchedReleaseFiles)
96c6cab1
MV
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
75ef8f14
MV
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;
c033d415
MV
951 unsigned long long ETA = 0;
952 if(CurrentCPS > 0)
953 ETA = (TotalBytes - CurrentBytes) / CurrentCPS;
75ef8f14 954
1e8b4c0f
MV
955 // only show the ETA if it makes sense
956 if (ETA > 0 && ETA < 172800 /* two days */ )
0c508b03 957 snprintf(msg,sizeof(msg), _("Retrieving file %li of %li (%s remaining)"), i, TotalItems, TimeToStr(ETA).c_str());
1e8b4c0f 958 else
0c508b03 959 snprintf(msg,sizeof(msg), _("Retrieving file %li of %li"), i, TotalItems);
1e8b4c0f 960
75ef8f14
MV
961 // build the status str
962 status << "dlstatus:" << i
96c6cab1 963 << ":" << std::setprecision(3) << Percent
d0cfa8ad
MV
964 << ":" << msg
965 << endl;
31bda500
DK
966
967 std::string const dlstatus = status.str();
d68d65ad 968 FileFd::Write(fd, dlstatus.c_str(), dlstatus.size());
75ef8f14
MV
969 }
970
024d1123 971 return true;
b98f2859
AL
972}
973 /*}}}*/
974// AcquireStatus::Start - Called when the download is started /*{{{*/
975// ---------------------------------------------------------------------
976/* We just reset the counters */
977void 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;
d568ed2d
AL
987 TotalItems = 0;
988 CurrentItems = 0;
b98f2859
AL
989}
990 /*}}}*/
a6568219 991// AcquireStatus::Stop - Finished downloading /*{{{*/
b98f2859
AL
992// ---------------------------------------------------------------------
993/* This accurately computes the elapsed time and the total overall CPS. */
994void pkgAcquireStatus::Stop()
995{
996 // Compute the CPS and elapsed time
997 struct timeval NewTime;
998 gettimeofday(&NewTime,0);
999
31a0531d
AL
1000 double Delta = NewTime.tv_sec - StartTime.tv_sec +
1001 (NewTime.tv_usec - StartTime.tv_usec)/1000000.0;
b98f2859 1002
b98f2859 1003 // Compute the CPS value
31a0531d 1004 if (Delta < 0.01)
e331f6ed
AL
1005 CurrentCPS = 0;
1006 else
31a0531d 1007 CurrentCPS = FetchedBytes/Delta;
b98f2859 1008 LastBytes = CurrentBytes;
dbbc5494 1009 ElapsedTime = (unsigned long long)Delta;
b98f2859
AL
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. */
73da43e9 1015void pkgAcquireStatus::Fetched(unsigned long long Size,unsigned long long Resume)
93274b8d 1016{
b98f2859
AL
1017 FetchedBytes += Size - Resume;
1018}
1019 /*}}}*/
862bafea 1020
9d653a6d
DK
1021APT_CONST pkgAcquire::UriIterator::~UriIterator() {}
1022APT_CONST pkgAcquire::MethodConfig::~MethodConfig() {}
1023APT_CONST pkgAcquireStatus::~pkgAcquireStatus() {}