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