merge from the lp:~mvo/apt/mvo branch
[ntk/apt.git] / apt-pkg / acquire-worker.cc
1 // -*- mode: cpp; mode: fold -*-
2 // Description /*{{{*/
3 // $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $
4 /* ######################################################################
5
6 Acquire Worker
7
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
11
12 ##################################################################### */
13 /*}}}*/
14 // Include Files /*{{{*/
15 #include <config.h>
16
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>
23
24 #include <iostream>
25 #include <sstream>
26 #include <fstream>
27
28 #include <sys/stat.h>
29 #include <unistd.h>
30 #include <fcntl.h>
31 #include <signal.h>
32 #include <stdio.h>
33 #include <errno.h>
34
35 #include <apti18n.h>
36 /*}}}*/
37
38 using namespace std;
39
40 // Worker::Worker - Constructor for Queue startup /*{{{*/
41 // ---------------------------------------------------------------------
42 /* */
43 pkgAcquire::Worker::Worker(Queue *Q,MethodConfig *Cnf,
44 pkgAcquireStatus *Log) : Log(Log)
45 {
46 OwnerQ = Q;
47 Config = Cnf;
48 Access = Cnf->Access;
49 CurrentItem = 0;
50 TotalSize = 0;
51 CurrentSize = 0;
52
53 Construct();
54 }
55 /*}}}*/
56 // Worker::Worker - Constructor for method config startup /*{{{*/
57 // ---------------------------------------------------------------------
58 /* */
59 pkgAcquire::Worker::Worker(MethodConfig *Cnf)
60 {
61 OwnerQ = 0;
62 Config = Cnf;
63 Access = Cnf->Access;
64 CurrentItem = 0;
65 TotalSize = 0;
66 CurrentSize = 0;
67
68 Construct();
69 }
70 /*}}}*/
71 // Worker::Construct - Constructor helper /*{{{*/
72 // ---------------------------------------------------------------------
73 /* */
74 void pkgAcquire::Worker::Construct()
75 {
76 NextQueue = 0;
77 NextAcquire = 0;
78 Process = -1;
79 InFd = -1;
80 OutFd = -1;
81 OutReady = false;
82 InReady = false;
83 Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
84 }
85 /*}}}*/
86 // Worker::~Worker - Destructor /*{{{*/
87 // ---------------------------------------------------------------------
88 /* */
89 pkgAcquire::Worker::~Worker()
90 {
91 close(InFd);
92 close(OutFd);
93
94 if (Process > 0)
95 {
96 /* Closing of stdin is the signal to exit and die when the process
97 indicates it needs cleanup */
98 if (Config->NeedsCleanup == false)
99 kill(Process,SIGINT);
100 ExecWait(Process,Access.c_str(),true);
101 }
102 }
103 /*}}}*/
104 // Worker::Start - Start the worker process /*{{{*/
105 // ---------------------------------------------------------------------
106 /* This forks the method and inits the communication channel */
107 bool pkgAcquire::Worker::Start()
108 {
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());
113
114 if (Debug == true)
115 clog << "Starting method '" << Method << '\'' << endl;
116
117 // Create the pipes
118 int Pipes[4] = {-1,-1,-1,-1};
119 if (pipe(Pipes) != 0 || pipe(Pipes+2) != 0)
120 {
121 _error->Errno("pipe","Failed to create IPC pipe to subprocess");
122 for (int I = 0; I != 4; I++)
123 close(Pipes[I]);
124 return false;
125 }
126 for (int I = 0; I != 4; I++)
127 SetCloseExec(Pipes[I],true);
128
129 // Fork off the process
130 Process = ExecFork();
131 if (Process == 0)
132 {
133 // Setup the FDs
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);
139
140 const char *Args[2];
141 Args[0] = Method.c_str();
142 Args[1] = 0;
143 execv(Args[0],(char **)Args);
144 cerr << "Failed to exec method " << Args[0] << endl;
145 _exit(100);
146 }
147
148 // Fix up our FDs
149 InFd = Pipes[0];
150 OutFd = Pipes[3];
151 SetNonBlock(Pipes[0],true);
152 SetNonBlock(Pipes[3],true);
153 close(Pipes[1]);
154 close(Pipes[2]);
155 OutReady = false;
156 InReady = true;
157
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());
162
163 RunMessages();
164 if (OwnerQ != 0)
165 SendConfiguration();
166
167 return true;
168 }
169 /*}}}*/
170 // Worker::ReadMessages - Read all pending messages into the list /*{{{*/
171 // ---------------------------------------------------------------------
172 /* */
173 bool pkgAcquire::Worker::ReadMessages()
174 {
175 if (::ReadMessages(InFd,MessageQueue) == false)
176 return MethodFailure();
177 return true;
178 }
179 /*}}}*/
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()
185 {
186 while (MessageQueue.empty() == false)
187 {
188 string Message = MessageQueue.front();
189 MessageQueue.erase(MessageQueue.begin());
190
191 if (Debug == true)
192 clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
193
194 // Fetch the message number
195 char *End;
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());
199
200 string URI = LookupTag(Message,"URI");
201 pkgAcquire::Queue::QItem *Itm = 0;
202 if (URI.empty() == false)
203 Itm = OwnerQ->FindItem(URI,this);
204
205 // update used mirror
206 string UsedMirror = LookupTag(Message,"UsedMirror", "");
207 if (!UsedMirror.empty() &&
208 Itm &&
209 Itm->Description.find(" ") != string::npos)
210 {
211 Itm->Description.replace(0, Itm->Description.find(" "), UsedMirror);
212 // FIXME: will we need this as well?
213 //Itm->ShortDesc = UsedMirror;
214 }
215
216 // Determine the message number and dispatch
217 switch (Number)
218 {
219 // 100 Capabilities
220 case 100:
221 if (Capabilities(Message) == false)
222 return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
223 break;
224
225 // 101 Log
226 case 101:
227 if (Debug == true)
228 clog << " <- (log) " << LookupTag(Message,"Message") << endl;
229 break;
230
231 // 102 Status
232 case 102:
233 Status = LookupTag(Message,"Message");
234 break;
235
236 // 103 Redirect
237 case 103:
238 {
239 if (Itm == 0)
240 {
241 _error->Error("Method gave invalid 103 Redirect message");
242 break;
243 }
244
245 string NewURI = LookupTag(Message,"New-URI",URI.c_str());
246 Itm->URI = NewURI;
247 break;
248 }
249
250 // 200 URI Start
251 case 200:
252 {
253 if (Itm == 0)
254 {
255 _error->Error("Method gave invalid 200 URI Start message");
256 break;
257 }
258
259 CurrentItem = Itm;
260 CurrentSize = 0;
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));
264
265 // Display update before completion
266 if (Log != 0 && Log->MorePulses == true)
267 Log->Pulse(Itm->Owner->GetOwner());
268
269 if (Log != 0)
270 Log->Fetch(*Itm);
271
272 break;
273 }
274
275 // 201 URI Done
276 case 201:
277 {
278 if (Itm == 0)
279 {
280 _error->Error("Method gave invalid 201 URI Done message");
281 break;
282 }
283
284 pkgAcquire::Item *Owner = Itm->Owner;
285 pkgAcquire::ItemDesc Desc = *Itm;
286
287 // Display update before completion
288 if (Log != 0 && Log->MorePulses == true)
289 Log->Pulse(Owner->GetOwner());
290
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);
296
297 // see if there is a hash to verify
298 string RecivedHash;
299 HashString expectedHash(Owner->HashSum());
300 if(!expectedHash.empty())
301 {
302 string hashTag = expectedHash.HashType()+"-Hash";
303 string hashSum = LookupTag(Message, hashTag.c_str());
304 if(!hashSum.empty())
305 RecivedHash = expectedHash.HashType() + ":" + hashSum;
306 if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
307 {
308 clog << "201 URI Done: " << Owner->DescURI() << endl
309 << "RecivedHash: " << RecivedHash << endl
310 << "ExpectedHash: " << expectedHash.toStr()
311 << endl << endl;
312 }
313 }
314 Owner->Done(Message, ServerSize, RecivedHash.c_str(), Config);
315 ItemDone();
316
317 // Log that we are done
318 if (Log != 0)
319 {
320 if (StringToBool(LookupTag(Message,"IMS-Hit"),false) == true ||
321 StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false) == true)
322 {
323 /* Hide 'hits' for local only sources - we also manage to
324 hide gets */
325 if (Config->LocalOnly == false)
326 Log->IMSHit(Desc);
327 }
328 else
329 Log->Done(Desc);
330 }
331 break;
332 }
333
334 // 400 URI Failure
335 case 400:
336 {
337 if (Itm == 0)
338 {
339 _error->Error("Method gave invalid 400 URI Failure message");
340 break;
341 }
342
343 // Display update before completion
344 if (Log != 0 && Log->MorePulses == true)
345 Log->Pulse(Itm->Owner->GetOwner());
346
347 pkgAcquire::Item *Owner = Itm->Owner;
348 pkgAcquire::ItemDesc Desc = *Itm;
349 OwnerQ->ItemDone(Itm);
350
351 // set some status
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;
357
358 Owner->Failed(Message,Config);
359 ItemDone();
360
361 if (Log != 0)
362 Log->Fail(Desc);
363
364 break;
365 }
366
367 // 401 General Failure
368 case 401:
369 _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
370 break;
371
372 // 403 Media Change
373 case 403:
374 MediaChange(Message);
375 break;
376 }
377 }
378 return true;
379 }
380 /*}}}*/
381 // Worker::Capabilities - 100 Capabilities handler /*{{{*/
382 // ---------------------------------------------------------------------
383 /* This parses the capabilities message and dumps it into the configuration
384 structure. */
385 bool pkgAcquire::Worker::Capabilities(string Message)
386 {
387 if (Config == 0)
388 return true;
389
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);
397
398 // Some debug text
399 if (Debug == true)
400 {
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;
409 }
410
411 return true;
412 }
413 /*}}}*/
414 // Worker::MediaChange - Request a media change /*{{{*/
415 // ---------------------------------------------------------------------
416 /* */
417 bool pkgAcquire::Worker::MediaChange(string Message)
418 {
419 int status_fd = _config->FindI("APT::Status-Fd",-1);
420 if(status_fd > 0)
421 {
422 string Media = LookupTag(Message,"Media");
423 string Drive = LookupTag(Message,"Drive");
424 ostringstream msg,status;
425 ioprintf(msg,_("Please insert the disc labeled: "
426 "'%s' "
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
433 << endl;
434
435 std::string const dlstatus = status.str();
436 FileFd::Write(status_fd, dlstatus.c_str(), dlstatus.size());
437 }
438
439 if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"),
440 LookupTag(Message,"Drive")) == false)
441 {
442 char S[300];
443 snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
444 if (Debug == true)
445 clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
446 OutQueue += S;
447 OutReady = true;
448 return true;
449 }
450
451 char S[300];
452 snprintf(S,sizeof(S),"603 Media Changed\n\n");
453 if (Debug == true)
454 clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
455 OutQueue += S;
456 OutReady = true;
457 return true;
458 }
459 /*}}}*/
460 // Worker::SendConfiguration - Send the config to the method /*{{{*/
461 // ---------------------------------------------------------------------
462 /* */
463 bool pkgAcquire::Worker::SendConfiguration()
464 {
465 if (Config->SendConfig == false)
466 return true;
467
468 if (OutFd == -1)
469 return false;
470
471 string Message = "601 Configuration\n";
472 Message.reserve(2000);
473
474 /* Write out all of the configuration directives by walking the
475 configuration tree */
476 const Configuration::Item *Top = _config->Tree(0);
477 for (; Top != 0;)
478 {
479 if (Top->Value.empty() == false)
480 {
481 string Line = "Config-Item: " + QuoteString(Top->FullTag(),"=\"\n") + "=";
482 Line += QuoteString(Top->Value,"\n") + '\n';
483 Message += Line;
484 }
485
486 if (Top->Child != 0)
487 {
488 Top = Top->Child;
489 continue;
490 }
491
492 while (Top != 0 && Top->Next == 0)
493 Top = Top->Parent;
494 if (Top != 0)
495 Top = Top->Next;
496 }
497 Message += '\n';
498
499 if (Debug == true)
500 clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
501 OutQueue += Message;
502 OutReady = true;
503
504 return true;
505 }
506 /*}}}*/
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)
511 {
512 if (OutFd == -1)
513 return false;
514
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();
520 Message += "\n\n";
521
522 if (Debug == true)
523 clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
524 OutQueue += Message;
525 OutReady = true;
526
527 return true;
528 }
529 /*}}}*/
530 // Worker::OutFdRead - Out bound FD is ready /*{{{*/
531 // ---------------------------------------------------------------------
532 /* */
533 bool pkgAcquire::Worker::OutFdReady()
534 {
535 if (FileFd::Write(OutFd,OutQueue.c_str(),OutQueue.length()) == false)
536 return MethodFailure();
537
538 OutQueue.clear();
539 if (OutQueue.empty() == true)
540 OutReady = false;
541
542 return true;
543 }
544 /*}}}*/
545 // Worker::InFdRead - In bound FD is ready /*{{{*/
546 // ---------------------------------------------------------------------
547 /* */
548 bool pkgAcquire::Worker::InFdReady()
549 {
550 if (ReadMessages() == false)
551 return false;
552 RunMessages();
553 return true;
554 }
555 /*}}}*/
556 // Worker::MethodFailure - Called when the method fails /*{{{*/
557 // ---------------------------------------------------------------------
558 /* This is called when the method is belived to have failed, probably because
559 read returned -1. */
560 bool pkgAcquire::Worker::MethodFailure()
561 {
562 _error->Error("Method %s has died unexpectedly!",Access.c_str());
563
564 // do not reap the child here to show meaningfull error to the user
565 ExecWait(Process,Access.c_str(),false);
566 Process = -1;
567 close(InFd);
568 close(OutFd);
569 InFd = -1;
570 OutFd = -1;
571 OutReady = false;
572 InReady = false;
573 OutQueue = string();
574 MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
575
576 return false;
577 }
578 /*}}}*/
579 // Worker::Pulse - Called periodically /*{{{*/
580 // ---------------------------------------------------------------------
581 /* */
582 void pkgAcquire::Worker::Pulse()
583 {
584 if (CurrentItem == 0)
585 return;
586
587 struct stat Buf;
588 if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
589 return;
590 CurrentSize = Buf.st_size;
591
592 // Hmm? Should not happen...
593 if (CurrentSize > TotalSize && TotalSize != 0)
594 TotalSize = CurrentSize;
595 }
596 /*}}}*/
597 // Worker::ItemDone - Called when the current item is finished /*{{{*/
598 // ---------------------------------------------------------------------
599 /* */
600 void pkgAcquire::Worker::ItemDone()
601 {
602 CurrentItem = 0;
603 CurrentSize = 0;
604 TotalSize = 0;
605 Status = string();
606 }
607 /*}}}*/