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