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