]>
git.saurik.com Git - apt-legacy.git/blob - apt-pkg/acquire.cc
1 // -*- mode: cpp; mode: fold -*-
3 // $Id: acquire.cc,v 1.50 2004/03/17 05:17:11 mdz Exp $
4 /* ######################################################################
6 Acquire - File Acquiration
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.
13 ##################################################################### */
15 // Include Files /*{{{*/
17 #pragma implementation "apt-pkg/acquire.h"
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 <apt-pkg/strutl.h>
39 // Acquire::pkgAcquire - Constructor /*{{{*/
40 // ---------------------------------------------------------------------
41 /* We grab some runtime state from the configuration space */
42 pkgAcquire::pkgAcquire(pkgAcquireStatus
*Log
) : Log(Log
)
50 string Mode
= _config
->Find("Acquire::Queue-Mode","host");
51 if (strcasecmp(Mode
.c_str(),"host") == 0)
52 QueueMode
= QueueHost
;
53 if (strcasecmp(Mode
.c_str(),"access") == 0)
54 QueueMode
= QueueAccess
;
56 Debug
= _config
->FindB("Debug::pkgAcquire",false);
58 // This is really a stupid place for this
60 if (stat((_config
->FindDir("Dir::State::lists") + "partial/").c_str(),&St
) != 0 ||
61 S_ISDIR(St
.st_mode
) == 0)
62 _error
->Error(_("Lists directory %spartial is missing."),
63 _config
->FindDir("Dir::State::lists").c_str());
64 if (stat((_config
->FindDir("Dir::Cache::Archives") + "partial/").c_str(),&St
) != 0 ||
65 S_ISDIR(St
.st_mode
) == 0)
66 _error
->Error(_("Archive directory %spartial is missing."),
67 _config
->FindDir("Dir::Cache::Archives").c_str());
70 // Acquire::~pkgAcquire - Destructor /*{{{*/
71 // ---------------------------------------------------------------------
72 /* Free our memory, clean up the queues (destroy the workers) */
73 pkgAcquire::~pkgAcquire()
79 MethodConfig
*Jnk
= Configs
;
80 Configs
= Configs
->Next
;
85 // Acquire::Shutdown - Clean out the acquire object /*{{{*/
86 // ---------------------------------------------------------------------
88 void pkgAcquire::Shutdown()
90 while (Items
.size() != 0)
92 if (Items
[0]->Status
== Item::StatFetching
)
93 Items
[0]->Status
= Item::StatError
;
100 Queues
= Queues
->Next
;
105 // Acquire::Add - Add a new item /*{{{*/
106 // ---------------------------------------------------------------------
107 /* This puts an item on the acquire list. This list is mainly for tracking
109 void pkgAcquire::Add(Item
*Itm
)
111 Items
.push_back(Itm
);
114 // Acquire::Remove - Remove a item /*{{{*/
115 // ---------------------------------------------------------------------
116 /* Remove an item from the acquire list. This is usually not used.. */
117 void pkgAcquire::Remove(Item
*Itm
)
121 for (ItemIterator I
= Items
.begin(); I
!= Items
.end();)
133 // Acquire::Add - Add a worker /*{{{*/
134 // ---------------------------------------------------------------------
135 /* A list of workers is kept so that the select loop can direct their FD
137 void pkgAcquire::Add(Worker
*Work
)
139 Work
->NextAcquire
= Workers
;
143 // Acquire::Remove - Remove a worker /*{{{*/
144 // ---------------------------------------------------------------------
145 /* A worker has died. This can not be done while the select loop is running
146 as it would require that RunFds could handling a changing list state and
148 void pkgAcquire::Remove(Worker
*Work
)
153 Worker
**I
= &Workers
;
157 *I
= (*I
)->NextAcquire
;
159 I
= &(*I
)->NextAcquire
;
163 // Acquire::Enqueue - Queue an URI for fetching /*{{{*/
164 // ---------------------------------------------------------------------
165 /* This is the entry point for an item. An item calls this function when
166 it is constructed which creates a queue (based on the current queue
167 mode) and puts the item in that queue. If the system is running then
168 the queue might be started. */
169 void pkgAcquire::Enqueue(ItemDesc
&Item
)
171 // Determine which queue to put the item in
172 const MethodConfig
*Config
;
173 string Name
= QueueName(Item
.URI
,Config
);
174 if (Name
.empty() == true)
177 // Find the queue structure
179 for (; I
!= 0 && I
->Name
!= Name
; I
= I
->Next
);
182 I
= new Queue(Name
,this);
190 // See if this is a local only URI
191 if (Config
->LocalOnly
== true && Item
.Owner
->Complete
== false)
192 Item
.Owner
->Local
= true;
193 Item
.Owner
->Status
= Item::StatIdle
;
195 // Queue it into the named queue
202 clog
<< "Fetching " << Item
.URI
<< endl
;
203 clog
<< " to " << Item
.Owner
->DestFile
<< endl
;
204 clog
<< " Queue is: " << Name
<< endl
;
208 // Acquire::Dequeue - Remove an item from all queues /*{{{*/
209 // ---------------------------------------------------------------------
210 /* This is called when an item is finished being fetched. It removes it
211 from all the queues */
212 void pkgAcquire::Dequeue(Item
*Itm
)
216 for (; I
!= 0; I
= I
->Next
)
217 Res
|= I
->Dequeue(Itm
);
220 clog
<< "Dequeuing " << Itm
->DestFile
<< endl
;
225 // Acquire::QueueName - Return the name of the queue for this URI /*{{{*/
226 // ---------------------------------------------------------------------
227 /* The string returned depends on the configuration settings and the
228 method parameters. Given something like http://foo.org/bar it can
229 return http://foo.org or http */
230 string
pkgAcquire::QueueName(string Uri
,MethodConfig
const *&Config
)
234 Config
= GetConfig(U
.Access
);
238 /* Single-Instance methods get exactly one queue per URI. This is
239 also used for the Access queue method */
240 if (Config
->SingleInstance
== true || QueueMode
== QueueAccess
)
243 return U
.Access
+ ':' + U
.Host
;
246 // Acquire::GetConfig - Fetch the configuration information /*{{{*/
247 // ---------------------------------------------------------------------
248 /* This locates the configuration structure for an access method. If
249 a config structure cannot be found a Worker will be created to
251 pkgAcquire::MethodConfig
*pkgAcquire::GetConfig(string Access
)
253 // Search for an existing config
255 for (Conf
= Configs
; Conf
!= 0; Conf
= Conf
->Next
)
256 if (Conf
->Access
== Access
)
259 // Create the new config class
260 Conf
= new MethodConfig
;
261 Conf
->Access
= Access
;
262 Conf
->Next
= Configs
;
265 // Create the worker to fetch the configuration
267 if (Work
.Start() == false)
270 /* if a method uses DownloadLimit, we switch to SingleInstance mode */
271 if(_config
->FindI("Acquire::"+Access
+"::DlLimit",0) > 0)
272 Conf
->SingleInstance
= true;
277 // Acquire::SetFds - Deal with readable FDs /*{{{*/
278 // ---------------------------------------------------------------------
279 /* Collect FDs that have activity monitors into the fd sets */
280 void pkgAcquire::SetFds(int &Fd
,fd_set
*RSet
,fd_set
*WSet
)
282 for (Worker
*I
= Workers
; I
!= 0; I
= I
->NextAcquire
)
284 if (I
->InReady
== true && I
->InFd
>= 0)
288 FD_SET(I
->InFd
,RSet
);
290 if (I
->OutReady
== true && I
->OutFd
>= 0)
294 FD_SET(I
->OutFd
,WSet
);
299 // Acquire::RunFds - Deal with active FDs /*{{{*/
300 // ---------------------------------------------------------------------
301 /* Dispatch active FDs over to the proper workers. It is very important
302 that a worker never be erased while this is running! The queue class
303 should never erase a worker except during shutdown processing. */
304 void pkgAcquire::RunFds(fd_set
*RSet
,fd_set
*WSet
)
306 for (Worker
*I
= Workers
; I
!= 0; I
= I
->NextAcquire
)
308 if (I
->InFd
>= 0 && FD_ISSET(I
->InFd
,RSet
) != 0)
310 if (I
->OutFd
>= 0 && FD_ISSET(I
->OutFd
,WSet
) != 0)
315 // Acquire::Run - Run the fetch sequence /*{{{*/
316 // ---------------------------------------------------------------------
317 /* This runs the queues. It manages a select loop for all of the
318 Worker tasks. The workers interact with the queues and items to
319 manage the actual fetch. */
320 pkgAcquire::RunResult
pkgAcquire::Run(int PulseIntervall
)
324 for (Queue
*I
= Queues
; I
!= 0; I
= I
->Next
)
330 bool WasCancelled
= false;
332 // Run till all things have been acquired
335 tv
.tv_usec
= PulseIntervall
;
343 SetFds(Highest
,&RFds
,&WFds
);
348 Res
= select(Highest
+1,&RFds
,&WFds
,0,&tv
);
350 while (Res
< 0 && errno
== EINTR
);
354 _error
->Errno("select","Select has failed");
359 if (_error
->PendingError() == true)
362 // Timeout, notify the log class
363 if (Res
== 0 || (Log
!= 0 && Log
->Update
== true))
365 tv
.tv_usec
= PulseIntervall
;
366 for (Worker
*I
= Workers
; I
!= 0; I
= I
->NextAcquire
)
368 if (Log
!= 0 && Log
->Pulse(this) == false)
379 // Shut down the acquire bits
381 for (Queue
*I
= Queues
; I
!= 0; I
= I
->Next
)
384 // Shut down the items
385 for (ItemIterator I
= Items
.begin(); I
!= Items
.end(); I
++)
388 if (_error
->PendingError())
395 // Acquire::Bump - Called when an item is dequeued /*{{{*/
396 // ---------------------------------------------------------------------
397 /* This routine bumps idle queues in hopes that they will be able to fetch
399 void pkgAcquire::Bump()
401 for (Queue
*I
= Queues
; I
!= 0; I
= I
->Next
)
405 // Acquire::WorkerStep - Step to the next worker /*{{{*/
406 // ---------------------------------------------------------------------
407 /* Not inlined to advoid including acquire-worker.h */
408 pkgAcquire::Worker
*pkgAcquire::WorkerStep(Worker
*I
)
410 return I
->NextAcquire
;
413 // Acquire::Clean - Cleans a directory /*{{{*/
414 // ---------------------------------------------------------------------
415 /* This is a bit simplistic, it looks at every file in the dir and sees
416 if it is part of the download set. */
417 bool pkgAcquire::Clean(string Dir
)
419 DIR *D
= opendir(Dir
.c_str());
421 return _error
->Errno("opendir",_("Unable to read %s"),Dir
.c_str());
423 string StartDir
= SafeGetCWD();
424 if (chdir(Dir
.c_str()) != 0)
427 return _error
->Errno("chdir",_("Unable to change to %s"),Dir
.c_str());
430 for (struct dirent
*Dir
= readdir(D
); Dir
!= 0; Dir
= readdir(D
))
433 if (strcmp(Dir
->d_name
,"lock") == 0 ||
434 strcmp(Dir
->d_name
,"partial") == 0 ||
435 strcmp(Dir
->d_name
,".") == 0 ||
436 strcmp(Dir
->d_name
,"..") == 0)
439 // Look in the get list
440 ItemCIterator I
= Items
.begin();
441 for (; I
!= Items
.end(); I
++)
442 if (flNotDir((*I
)->DestFile
) == Dir
->d_name
)
445 // Nothing found, nuke it
446 if (I
== Items
.end())
450 chdir(StartDir
.c_str());
455 // Acquire::TotalNeeded - Number of bytes to fetch /*{{{*/
456 // ---------------------------------------------------------------------
457 /* This is the total number of bytes needed */
458 double pkgAcquire::TotalNeeded()
461 for (ItemCIterator I
= ItemsBegin(); I
!= ItemsEnd(); I
++)
462 Total
+= (*I
)->FileSize
;
466 // Acquire::FetchNeeded - Number of bytes needed to get /*{{{*/
467 // ---------------------------------------------------------------------
468 /* This is the number of bytes that is not local */
469 double pkgAcquire::FetchNeeded()
472 for (ItemCIterator I
= ItemsBegin(); I
!= ItemsEnd(); I
++)
473 if ((*I
)->Local
== false)
474 Total
+= (*I
)->FileSize
;
478 // Acquire::PartialPresent - Number of partial bytes we already have /*{{{*/
479 // ---------------------------------------------------------------------
480 /* This is the number of bytes that is not local */
481 double pkgAcquire::PartialPresent()
484 for (ItemCIterator I
= ItemsBegin(); I
!= ItemsEnd(); I
++)
485 if ((*I
)->Local
== false)
486 Total
+= (*I
)->PartialSize
;
490 // Acquire::UriBegin - Start iterator for the uri list /*{{{*/
491 // ---------------------------------------------------------------------
493 pkgAcquire::UriIterator
pkgAcquire::UriBegin()
495 return UriIterator(Queues
);
498 // Acquire::UriEnd - End iterator for the uri list /*{{{*/
499 // ---------------------------------------------------------------------
501 pkgAcquire::UriIterator
pkgAcquire::UriEnd()
503 return UriIterator(0);
507 // Acquire::MethodConfig::MethodConfig - Constructor /*{{{*/
508 // ---------------------------------------------------------------------
510 pkgAcquire::MethodConfig::MethodConfig()
512 SingleInstance
= false;
521 // Queue::Queue - Constructor /*{{{*/
522 // ---------------------------------------------------------------------
524 pkgAcquire::Queue::Queue(string Name
,pkgAcquire
*Owner
) : Name(Name
),
534 // Queue::~Queue - Destructor /*{{{*/
535 // ---------------------------------------------------------------------
537 pkgAcquire::Queue::~Queue()
549 // Queue::Enqueue - Queue an item to the queue /*{{{*/
550 // ---------------------------------------------------------------------
552 void pkgAcquire::Queue::Enqueue(ItemDesc
&Item
)
555 for (; *I
!= 0; I
= &(*I
)->Next
);
558 QItem
*Itm
= new QItem
;
563 Item
.Owner
->QueueCounter
++;
564 if (Items
->Next
== 0)
568 // Queue::Dequeue - Remove an item from the queue /*{{{*/
569 // ---------------------------------------------------------------------
570 /* We return true if we hit something */
571 bool pkgAcquire::Queue::Dequeue(Item
*Owner
)
573 if (Owner
->Status
== pkgAcquire::Item::StatFetching
)
574 return _error
->Error("Tried to dequeue a fetching object");
581 if ((*I
)->Owner
== Owner
)
585 Owner
->QueueCounter
--;
596 // Queue::Startup - Start the worker processes /*{{{*/
597 // ---------------------------------------------------------------------
598 /* It is possible for this to be called with a pre-existing set of
600 bool pkgAcquire::Queue::Startup()
605 pkgAcquire::MethodConfig
*Cnf
= Owner
->GetConfig(U
.Access
);
609 Workers
= new Worker(this,Cnf
,Owner
->Log
);
611 if (Workers
->Start() == false)
614 /* When pipelining we commit 10 items. This needs to change when we
615 added other source retry to have cycle maintain a pipeline depth
617 if (Cnf
->Pipeline
== true)
626 // Queue::Shutdown - Shutdown the worker processes /*{{{*/
627 // ---------------------------------------------------------------------
628 /* If final is true then all workers are eliminated, otherwise only workers
629 that do not need cleanup are removed */
630 bool pkgAcquire::Queue::Shutdown(bool Final
)
632 // Delete all of the workers
633 pkgAcquire::Worker
**Cur
= &Workers
;
636 pkgAcquire::Worker
*Jnk
= *Cur
;
637 if (Final
== true || Jnk
->GetConf()->NeedsCleanup
== false)
639 *Cur
= Jnk
->NextQueue
;
644 Cur
= &(*Cur
)->NextQueue
;
650 // Queue::FindItem - Find a URI in the item list /*{{{*/
651 // ---------------------------------------------------------------------
653 pkgAcquire::Queue::QItem
*pkgAcquire::Queue::FindItem(string URI
,pkgAcquire::Worker
*Owner
)
655 for (QItem
*I
= Items
; I
!= 0; I
= I
->Next
)
656 if (I
->URI
== URI
&& I
->Worker
== Owner
)
661 // Queue::ItemDone - Item has been completed /*{{{*/
662 // ---------------------------------------------------------------------
663 /* The worker signals this which causes the item to be removed from the
664 queue. If this is the last queue instance then it is removed from the
666 bool pkgAcquire::Queue::ItemDone(QItem
*Itm
)
669 if (Itm
->Owner
->Status
== pkgAcquire::Item::StatFetching
)
670 Itm
->Owner
->Status
= pkgAcquire::Item::StatDone
;
672 if (Itm
->Owner
->QueueCounter
<= 1)
673 Owner
->Dequeue(Itm
->Owner
);
683 // Queue::Cycle - Queue new items into the method /*{{{*/
684 // ---------------------------------------------------------------------
685 /* This locates a new idle item and sends it to the worker. If pipelining
686 is enabled then it keeps the pipe full. */
687 bool pkgAcquire::Queue::Cycle()
689 if (Items
== 0 || Workers
== 0)
693 return _error
->Error("Pipedepth failure");
695 // Look for a queable item
697 while (PipeDepth
< (signed)MaxPipeDepth
)
699 for (; I
!= 0; I
= I
->Next
)
700 if (I
->Owner
->Status
== pkgAcquire::Item::StatIdle
)
703 // Nothing to do, queue is idle.
708 I
->Owner
->Status
= pkgAcquire::Item::StatFetching
;
710 if (Workers
->QueueItem(I
) == false)
717 // Queue::Bump - Fetch any pending objects if we are idle /*{{{*/
718 // ---------------------------------------------------------------------
719 /* This is called when an item in multiple queues is dequeued */
720 void pkgAcquire::Queue::Bump()
726 // AcquireStatus::pkgAcquireStatus - Constructor /*{{{*/
727 // ---------------------------------------------------------------------
729 pkgAcquireStatus::pkgAcquireStatus() : Update(true), MorePulses(false)
734 // AcquireStatus::Pulse - Called periodically /*{{{*/
735 // ---------------------------------------------------------------------
736 /* This computes some internal state variables for the derived classes to
737 use. It generates the current downloaded bytes and total bytes to download
738 as well as the current CPS estimate. */
739 bool pkgAcquireStatus::Pulse(pkgAcquire
*Owner
)
746 // Compute the total number of bytes to fetch
747 unsigned int Unknown
= 0;
748 unsigned int Count
= 0;
749 for (pkgAcquire::ItemCIterator I
= Owner
->ItemsBegin(); I
!= Owner
->ItemsEnd();
753 if ((*I
)->Status
== pkgAcquire::Item::StatDone
)
756 // Totally ignore local items
757 if ((*I
)->Local
== true)
760 TotalBytes
+= (*I
)->FileSize
;
761 if ((*I
)->Complete
== true)
762 CurrentBytes
+= (*I
)->FileSize
;
763 if ((*I
)->FileSize
== 0 && (*I
)->Complete
== false)
767 // Compute the current completion
768 unsigned long ResumeSize
= 0;
769 for (pkgAcquire::Worker
*I
= Owner
->WorkersBegin(); I
!= 0;
770 I
= Owner
->WorkerStep(I
))
771 if (I
->CurrentItem
!= 0 && I
->CurrentItem
->Owner
->Complete
== false)
773 CurrentBytes
+= I
->CurrentSize
;
774 ResumeSize
+= I
->ResumePoint
;
776 // Files with unknown size always have 100% completion
777 if (I
->CurrentItem
->Owner
->FileSize
== 0 &&
778 I
->CurrentItem
->Owner
->Complete
== false)
779 TotalBytes
+= I
->CurrentSize
;
782 // Normalize the figures and account for unknown size downloads
785 if (Unknown
== Count
)
786 TotalBytes
= Unknown
;
788 // Wha?! Is not supposed to happen.
789 if (CurrentBytes
> TotalBytes
)
790 CurrentBytes
= TotalBytes
;
793 struct timeval NewTime
;
794 gettimeofday(&NewTime
,0);
795 if (NewTime
.tv_sec
- Time
.tv_sec
== 6 && NewTime
.tv_usec
> Time
.tv_usec
||
796 NewTime
.tv_sec
- Time
.tv_sec
> 6)
798 double Delta
= NewTime
.tv_sec
- Time
.tv_sec
+
799 (NewTime
.tv_usec
- Time
.tv_usec
)/1000000.0;
801 // Compute the CPS value
805 CurrentCPS
= ((CurrentBytes
- ResumeSize
) - LastBytes
)/Delta
;
806 LastBytes
= CurrentBytes
- ResumeSize
;
807 ElapsedTime
= (unsigned long)Delta
;
811 int fd
= _config
->FindI("APT::Status-Fd",-1);
814 ostringstream status
;
817 long i
= CurrentItems
< TotalItems
? CurrentItems
+ 1 : CurrentItems
;
819 (unsigned long)((TotalBytes
- CurrentBytes
) / CurrentCPS
);
821 // only show the ETA if it makes sense
822 if (ETA
> 0 && ETA
< 172800 /* two days */ )
823 snprintf(msg
,sizeof(msg
), _("Retrieving file %li of %li (%s remaining)"), i
, TotalItems
, TimeToStr(ETA
).c_str());
825 snprintf(msg
,sizeof(msg
), _("Retrieving file %li of %li"), i
, TotalItems
);
829 // build the status str
830 status
<< "dlstatus:" << i
831 << ":" << (CurrentBytes
/float(TotalBytes
)*100.0)
834 write(fd
, status
.str().c_str(), status
.str().size());
840 // AcquireStatus::Start - Called when the download is started /*{{{*/
841 // ---------------------------------------------------------------------
842 /* We just reset the counters */
843 void pkgAcquireStatus::Start()
845 gettimeofday(&Time
,0);
846 gettimeofday(&StartTime
,0);
857 // AcquireStatus::Stop - Finished downloading /*{{{*/
858 // ---------------------------------------------------------------------
859 /* This accurately computes the elapsed time and the total overall CPS. */
860 void pkgAcquireStatus::Stop()
862 // Compute the CPS and elapsed time
863 struct timeval NewTime
;
864 gettimeofday(&NewTime
,0);
866 double Delta
= NewTime
.tv_sec
- StartTime
.tv_sec
+
867 (NewTime
.tv_usec
- StartTime
.tv_usec
)/1000000.0;
869 // Compute the CPS value
873 CurrentCPS
= FetchedBytes
/Delta
;
874 LastBytes
= CurrentBytes
;
875 ElapsedTime
= (unsigned int)Delta
;
878 // AcquireStatus::Fetched - Called when a byte set has been fetched /*{{{*/
879 // ---------------------------------------------------------------------
880 /* This is used to get accurate final transfer rate reporting. */
881 void pkgAcquireStatus::Fetched(unsigned long Size
,unsigned long Resume
)
883 FetchedBytes
+= Size
- Resume
;