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