1 // -*- mode: cpp; mode: fold -*-
3 // $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $
4 /* ######################################################################
8 The worker process can startup either as a Configuration prober
9 or as a queue runner. As a configuration prober it only reads the
10 configuration message and
12 ##################################################################### */
14 // Include Files /*{{{*/
17 #include <apt-pkg/acquire-worker.h>
18 #include <apt-pkg/acquire-item.h>
19 #include <apt-pkg/configuration.h>
20 #include <apt-pkg/error.h>
21 #include <apt-pkg/fileutl.h>
22 #include <apt-pkg/strutl.h>
40 // Worker::Worker - Constructor for Queue startup /*{{{*/
41 // ---------------------------------------------------------------------
43 pkgAcquire::Worker::Worker(Queue
*Q
,MethodConfig
*Cnf
,
44 pkgAcquireStatus
*Log
) : Log(Log
)
56 // Worker::Worker - Constructor for method config startup /*{{{*/
57 // ---------------------------------------------------------------------
59 pkgAcquire::Worker::Worker(MethodConfig
*Cnf
)
71 // Worker::Construct - Constructor helper /*{{{*/
72 // ---------------------------------------------------------------------
74 void pkgAcquire::Worker::Construct()
83 Debug
= _config
->FindB("Debug::pkgAcquire::Worker",false);
86 // Worker::~Worker - Destructor /*{{{*/
87 // ---------------------------------------------------------------------
89 pkgAcquire::Worker::~Worker()
96 /* Closing of stdin is the signal to exit and die when the process
97 indicates it needs cleanup */
98 if (Config
->NeedsCleanup
== false)
100 ExecWait(Process
,Access
.c_str(),true);
104 // Worker::Start - Start the worker process /*{{{*/
105 // ---------------------------------------------------------------------
106 /* This forks the method and inits the communication channel */
107 bool pkgAcquire::Worker::Start()
109 // Get the method path
110 string Method
= _config
->FindDir("Dir::Bin::Methods") + Access
;
111 if (FileExists(Method
) == false)
112 return _error
->Error(_("The method driver %s could not be found."),Method
.c_str());
115 clog
<< "Starting method '" << Method
<< '\'' << endl
;
118 int Pipes
[4] = {-1,-1,-1,-1};
119 if (pipe(Pipes
) != 0 || pipe(Pipes
+2) != 0)
121 _error
->Errno("pipe","Failed to create IPC pipe to subprocess");
122 for (int I
= 0; I
!= 4; I
++)
126 for (int I
= 0; I
!= 4; I
++)
127 SetCloseExec(Pipes
[I
],true);
129 // Fork off the process
130 Process
= ExecFork();
134 dup2(Pipes
[1],STDOUT_FILENO
);
135 dup2(Pipes
[2],STDIN_FILENO
);
136 SetCloseExec(STDOUT_FILENO
,false);
137 SetCloseExec(STDIN_FILENO
,false);
138 SetCloseExec(STDERR_FILENO
,false);
141 Args
[0] = Method
.c_str();
143 execv(Args
[0],(char **)Args
);
144 cerr
<< "Failed to exec method " << Args
[0] << endl
;
151 SetNonBlock(Pipes
[0],true);
152 SetNonBlock(Pipes
[3],true);
158 // Read the configuration data
159 if (WaitFd(InFd
) == false ||
160 ReadMessages() == false)
161 return _error
->Error(_("Method %s did not start correctly"),Method
.c_str());
170 // Worker::ReadMessages - Read all pending messages into the list /*{{{*/
171 // ---------------------------------------------------------------------
173 bool pkgAcquire::Worker::ReadMessages()
175 if (::ReadMessages(InFd
,MessageQueue
) == false)
176 return MethodFailure();
180 // Worker::RunMessage - Empty the message queue /*{{{*/
181 // ---------------------------------------------------------------------
182 /* This takes the messages from the message queue and runs them through
183 the parsers in order. */
184 bool pkgAcquire::Worker::RunMessages()
186 while (MessageQueue
.empty() == false)
188 string Message
= MessageQueue
.front();
189 MessageQueue
.erase(MessageQueue
.begin());
192 clog
<< " <- " << Access
<< ':' << QuoteString(Message
,"\n") << endl
;
194 // Fetch the message number
196 int Number
= strtol(Message
.c_str(),&End
,10);
197 if (End
== Message
.c_str())
198 return _error
->Error("Invalid message from method %s: %s",Access
.c_str(),Message
.c_str());
200 string URI
= LookupTag(Message
,"URI");
201 pkgAcquire::Queue::QItem
*Itm
= 0;
202 if (URI
.empty() == false)
203 Itm
= OwnerQ
->FindItem(URI
,this);
205 // update used mirror
206 string UsedMirror
= LookupTag(Message
,"UsedMirror", "");
207 if (!UsedMirror
.empty() &&
209 Itm
->Description
.find(" ") != string::npos
)
211 Itm
->Description
.replace(0, Itm
->Description
.find(" "), UsedMirror
);
212 // FIXME: will we need this as well?
213 //Itm->ShortDesc = UsedMirror;
216 // Determine the message number and dispatch
221 if (Capabilities(Message
) == false)
222 return _error
->Error("Unable to process Capabilities message from %s",Access
.c_str());
228 clog
<< " <- (log) " << LookupTag(Message
,"Message") << endl
;
233 Status
= LookupTag(Message
,"Message");
241 _error
->Error("Method gave invalid 103 Redirect message");
245 string NewURI
= LookupTag(Message
,"New-URI",URI
.c_str());
255 _error
->Error("Method gave invalid 200 URI Start message");
261 TotalSize
= strtoull(LookupTag(Message
,"Size","0").c_str(), NULL
, 10);
262 ResumePoint
= strtoull(LookupTag(Message
,"Resume-Point","0").c_str(), NULL
, 10);
263 Itm
->Owner
->Start(Message
,strtoull(LookupTag(Message
,"Size","0").c_str(), NULL
, 10));
265 // Display update before completion
266 if (Log
!= 0 && Log
->MorePulses
== true)
267 Log
->Pulse(Itm
->Owner
->GetOwner());
280 _error
->Error("Method gave invalid 201 URI Done message");
284 pkgAcquire::Item
*Owner
= Itm
->Owner
;
285 pkgAcquire::ItemDesc Desc
= *Itm
;
287 // Display update before completion
288 if (Log
!= 0 && Log
->MorePulses
== true)
289 Log
->Pulse(Owner
->GetOwner());
291 OwnerQ
->ItemDone(Itm
);
292 unsigned long long const ServerSize
= strtoull(LookupTag(Message
,"Size","0").c_str(), NULL
, 10);
293 if (TotalSize
!= 0 && ServerSize
!= TotalSize
)
294 _error
->Warning("Size of file %s is not what the server reported %s %llu",
295 Owner
->DestFile
.c_str(), LookupTag(Message
,"Size","0").c_str(),TotalSize
);
297 // see if there is a hash to verify
299 HashString
expectedHash(Owner
->HashSum());
300 if(!expectedHash
.empty())
302 string hashTag
= expectedHash
.HashType()+"-Hash";
303 string hashSum
= LookupTag(Message
, hashTag
.c_str());
305 RecivedHash
= expectedHash
.HashType() + ":" + hashSum
;
306 if(_config
->FindB("Debug::pkgAcquire::Auth", false) == true)
308 clog
<< "201 URI Done: " << Owner
->DescURI() << endl
309 << "RecivedHash: " << RecivedHash
<< endl
310 << "ExpectedHash: " << expectedHash
.toStr()
314 Owner
->Done(Message
, ServerSize
, RecivedHash
.c_str(), Config
);
317 // Log that we are done
320 if (StringToBool(LookupTag(Message
,"IMS-Hit"),false) == true ||
321 StringToBool(LookupTag(Message
,"Alt-IMS-Hit"),false) == true)
323 /* Hide 'hits' for local only sources - we also manage to
325 if (Config
->LocalOnly
== false)
339 _error
->Error("Method gave invalid 400 URI Failure message");
343 // Display update before completion
344 if (Log
!= 0 && Log
->MorePulses
== true)
345 Log
->Pulse(Itm
->Owner
->GetOwner());
347 pkgAcquire::Item
*Owner
= Itm
->Owner
;
348 pkgAcquire::ItemDesc Desc
= *Itm
;
349 OwnerQ
->ItemDone(Itm
);
352 if(LookupTag(Message
,"FailReason") == "Timeout" ||
353 LookupTag(Message
,"FailReason") == "TmpResolveFailure" ||
354 LookupTag(Message
,"FailReason") == "ResolveFailure" ||
355 LookupTag(Message
,"FailReason") == "ConnectionRefused")
356 Owner
->Status
= pkgAcquire::Item::StatTransientNetworkError
;
358 Owner
->Failed(Message
,Config
);
367 // 401 General Failure
369 _error
->Error("Method %s General failure: %s",Access
.c_str(),LookupTag(Message
,"Message").c_str());
374 MediaChange(Message
);
381 // Worker::Capabilities - 100 Capabilities handler /*{{{*/
382 // ---------------------------------------------------------------------
383 /* This parses the capabilities message and dumps it into the configuration
385 bool pkgAcquire::Worker::Capabilities(string Message
)
390 Config
->Version
= LookupTag(Message
,"Version");
391 Config
->SingleInstance
= StringToBool(LookupTag(Message
,"Single-Instance"),false);
392 Config
->Pipeline
= StringToBool(LookupTag(Message
,"Pipeline"),false);
393 Config
->SendConfig
= StringToBool(LookupTag(Message
,"Send-Config"),false);
394 Config
->LocalOnly
= StringToBool(LookupTag(Message
,"Local-Only"),false);
395 Config
->NeedsCleanup
= StringToBool(LookupTag(Message
,"Needs-Cleanup"),false);
396 Config
->Removable
= StringToBool(LookupTag(Message
,"Removable"),false);
401 clog
<< "Configured access method " << Config
->Access
<< endl
;
402 clog
<< "Version:" << Config
->Version
<<
403 " SingleInstance:" << Config
->SingleInstance
<<
404 " Pipeline:" << Config
->Pipeline
<<
405 " SendConfig:" << Config
->SendConfig
<<
406 " LocalOnly: " << Config
->LocalOnly
<<
407 " NeedsCleanup: " << Config
->NeedsCleanup
<<
408 " Removable: " << Config
->Removable
<< endl
;
414 // Worker::MediaChange - Request a media change /*{{{*/
415 // ---------------------------------------------------------------------
417 bool pkgAcquire::Worker::MediaChange(string Message
)
419 int status_fd
= _config
->FindI("APT::Status-Fd",-1);
422 string Media
= LookupTag(Message
,"Media");
423 string Drive
= LookupTag(Message
,"Drive");
424 ostringstream msg
,status
;
425 ioprintf(msg
,_("Please insert the disc labeled: "
427 "in the drive '%s' and press enter."),
428 Media
.c_str(),Drive
.c_str());
429 status
<< "media-change: " // message
430 << Media
<< ":" // media
431 << Drive
<< ":" // drive
432 << msg
.str() // l10n message
435 std::string
const dlstatus
= status
.str();
436 FileFd::Write(status_fd
, dlstatus
.c_str(), dlstatus
.size());
439 if (Log
== 0 || Log
->MediaChange(LookupTag(Message
,"Media"),
440 LookupTag(Message
,"Drive")) == false)
443 snprintf(S
,sizeof(S
),"603 Media Changed\nFailed: true\n\n");
445 clog
<< " -> " << Access
<< ':' << QuoteString(S
,"\n") << endl
;
452 snprintf(S
,sizeof(S
),"603 Media Changed\n\n");
454 clog
<< " -> " << Access
<< ':' << QuoteString(S
,"\n") << endl
;
460 // Worker::SendConfiguration - Send the config to the method /*{{{*/
461 // ---------------------------------------------------------------------
463 bool pkgAcquire::Worker::SendConfiguration()
465 if (Config
->SendConfig
== false)
471 string Message
= "601 Configuration\n";
472 Message
.reserve(2000);
474 /* Write out all of the configuration directives by walking the
475 configuration tree */
476 const Configuration::Item
*Top
= _config
->Tree(0);
479 if (Top
->Value
.empty() == false)
481 string Line
= "Config-Item: " + QuoteString(Top
->FullTag(),"=\"\n") + "=";
482 Line
+= QuoteString(Top
->Value
,"\n") + '\n';
492 while (Top
!= 0 && Top
->Next
== 0)
500 clog
<< " -> " << Access
<< ':' << QuoteString(Message
,"\n") << endl
;
507 // Worker::QueueItem - Add an item to the outbound queue /*{{{*/
508 // ---------------------------------------------------------------------
509 /* Send a URI Acquire message to the method */
510 bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem
*Item
)
515 string Message
= "600 URI Acquire\n";
516 Message
.reserve(300);
517 Message
+= "URI: " + Item
->URI
;
518 Message
+= "\nFilename: " + Item
->Owner
->DestFile
;
519 Message
+= Item
->Owner
->Custom600Headers();
523 clog
<< " -> " << Access
<< ':' << QuoteString(Message
,"\n") << endl
;
530 // Worker::OutFdRead - Out bound FD is ready /*{{{*/
531 // ---------------------------------------------------------------------
533 bool pkgAcquire::Worker::OutFdReady()
535 if (FileFd::Write(OutFd
,OutQueue
.c_str(),OutQueue
.length()) == false)
536 return MethodFailure();
539 if (OutQueue
.empty() == true)
545 // Worker::InFdRead - In bound FD is ready /*{{{*/
546 // ---------------------------------------------------------------------
548 bool pkgAcquire::Worker::InFdReady()
550 if (ReadMessages() == false)
556 // Worker::MethodFailure - Called when the method fails /*{{{*/
557 // ---------------------------------------------------------------------
558 /* This is called when the method is belived to have failed, probably because
560 bool pkgAcquire::Worker::MethodFailure()
562 _error
->Error("Method %s has died unexpectedly!",Access
.c_str());
564 // do not reap the child here to show meaningfull error to the user
565 ExecWait(Process
,Access
.c_str(),false);
574 MessageQueue
.erase(MessageQueue
.begin(),MessageQueue
.end());
579 // Worker::Pulse - Called periodically /*{{{*/
580 // ---------------------------------------------------------------------
582 void pkgAcquire::Worker::Pulse()
584 if (CurrentItem
== 0)
588 if (stat(CurrentItem
->Owner
->DestFile
.c_str(),&Buf
) != 0)
590 CurrentSize
= Buf
.st_size
;
592 // Hmm? Should not happen...
593 if (CurrentSize
> TotalSize
&& TotalSize
!= 0)
594 TotalSize
= CurrentSize
;
597 // Worker::ItemDone - Called when the current item is finished /*{{{*/
598 // ---------------------------------------------------------------------
600 void pkgAcquire::Worker::ItemDone()