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