]> git.saurik.com Git - apt.git/blob - apt-pkg/acquire-worker.cc
* apt-pkg/depcache.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 // Determine the message number and dispatch
204 switch (Number)
205 {
206 // 100 Capabilities
207 case 100:
208 if (Capabilities(Message) == false)
209 return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
210 break;
211
212 // 101 Log
213 case 101:
214 if (Debug == true)
215 clog << " <- (log) " << LookupTag(Message,"Message") << endl;
216 break;
217
218 // 102 Status
219 case 102:
220 Status = LookupTag(Message,"Message");
221 break;
222
223 // 103 Redirect
224 case 103:
225 {
226 if (Itm == 0)
227 {
228 _error->Error("Method gave invalid 103 Redirect message");
229 break;
230 }
231
232 string NewURI = LookupTag(Message,"New-URI",URI.c_str());
233 Itm->URI = NewURI;
234 break;
235 }
236
237 // 200 URI Start
238 case 200:
239 {
240 if (Itm == 0)
241 {
242 _error->Error("Method gave invalid 200 URI Start message");
243 break;
244 }
245
246 CurrentItem = Itm;
247 CurrentSize = 0;
248 TotalSize = atoi(LookupTag(Message,"Size","0").c_str());
249 ResumePoint = atoi(LookupTag(Message,"Resume-Point","0").c_str());
250 Itm->Owner->Start(Message,atoi(LookupTag(Message,"Size","0").c_str()));
251
252 // Display update before completion
253 if (Log != 0 && Log->MorePulses == true)
254 Log->Pulse(Itm->Owner->GetOwner());
255
256 if (Log != 0)
257 Log->Fetch(*Itm);
258
259 break;
260 }
261
262 // 201 URI Done
263 case 201:
264 {
265 if (Itm == 0)
266 {
267 _error->Error("Method gave invalid 201 URI Done message");
268 break;
269 }
270
271 pkgAcquire::Item *Owner = Itm->Owner;
272 pkgAcquire::ItemDesc Desc = *Itm;
273
274 // Display update before completion
275 if (Log != 0 && Log->MorePulses == true)
276 Log->Pulse(Owner->GetOwner());
277
278 OwnerQ->ItemDone(Itm);
279 if (TotalSize != 0 &&
280 (unsigned)atoi(LookupTag(Message,"Size","0").c_str()) != TotalSize)
281 _error->Warning("Bizarre Error - File size is not what the server reported %s %lu",
282 LookupTag(Message,"Size","0").c_str(),TotalSize);
283
284 // see if there is a hash to verify
285 string RecivedHash;
286 HashString expectedHash(Owner->HashSum());
287 if(!expectedHash.empty())
288 {
289 string hashTag = expectedHash.HashType()+"-Hash";
290 string hashSum = LookupTag(Message, hashTag.c_str());
291 if(!hashSum.empty())
292 RecivedHash = expectedHash.HashType() + ":" + hashSum;
293 if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
294 {
295 clog << "201 URI Done: " << Owner->DescURI() << endl
296 << "RecivedHash: " << RecivedHash << endl
297 << "ExpectedHash: " << expectedHash.toStr()
298 << endl << endl;
299 }
300 }
301 Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
302 RecivedHash.c_str(), Config);
303 ItemDone();
304
305 // Log that we are done
306 if (Log != 0)
307 {
308 if (StringToBool(LookupTag(Message,"IMS-Hit"),false) == true ||
309 StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false) == true)
310 {
311 /* Hide 'hits' for local only sources - we also manage to
312 hide gets */
313 if (Config->LocalOnly == false)
314 Log->IMSHit(Desc);
315 }
316 else
317 Log->Done(Desc);
318 }
319 break;
320 }
321
322 // 400 URI Failure
323 case 400:
324 {
325 if (Itm == 0)
326 {
327 _error->Error("Method gave invalid 400 URI Failure message");
328 break;
329 }
330
331 // Display update before completion
332 if (Log != 0 && Log->MorePulses == true)
333 Log->Pulse(Itm->Owner->GetOwner());
334
335 pkgAcquire::Item *Owner = Itm->Owner;
336 pkgAcquire::ItemDesc Desc = *Itm;
337 OwnerQ->ItemDone(Itm);
338
339 // set some status
340 if(LookupTag(Message,"FailReason") == "Timeout" ||
341 LookupTag(Message,"FailReason") == "TmpResolveFailure" ||
342 LookupTag(Message,"FailReason") == "ResolveFailure" ||
343 LookupTag(Message,"FailReason") == "ConnectionRefused")
344 Owner->Status = pkgAcquire::Item::StatTransientNetworkError;
345
346 Owner->Failed(Message,Config);
347 ItemDone();
348
349 if (Log != 0)
350 Log->Fail(Desc);
351
352 break;
353 }
354
355 // 401 General Failure
356 case 401:
357 _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
358 break;
359
360 // 403 Media Change
361 case 403:
362 MediaChange(Message);
363 break;
364 }
365 }
366 return true;
367 }
368 /*}}}*/
369 // Worker::Capabilities - 100 Capabilities handler /*{{{*/
370 // ---------------------------------------------------------------------
371 /* This parses the capabilities message and dumps it into the configuration
372 structure. */
373 bool pkgAcquire::Worker::Capabilities(string Message)
374 {
375 if (Config == 0)
376 return true;
377
378 Config->Version = LookupTag(Message,"Version");
379 Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
380 Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
381 Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
382 Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
383 Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
384 Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);
385
386 // Some debug text
387 if (Debug == true)
388 {
389 clog << "Configured access method " << Config->Access << endl;
390 clog << "Version:" << Config->Version <<
391 " SingleInstance:" << Config->SingleInstance <<
392 " Pipeline:" << Config->Pipeline <<
393 " SendConfig:" << Config->SendConfig <<
394 " LocalOnly: " << Config->LocalOnly <<
395 " NeedsCleanup: " << Config->NeedsCleanup <<
396 " Removable: " << Config->Removable << endl;
397 }
398
399 return true;
400 }
401 /*}}}*/
402 // Worker::MediaChange - Request a media change /*{{{*/
403 // ---------------------------------------------------------------------
404 /* */
405 bool pkgAcquire::Worker::MediaChange(string Message)
406 {
407 int status_fd = _config->FindI("APT::Status-Fd",-1);
408 if(status_fd > 0)
409 {
410 string Media = LookupTag(Message,"Media");
411 string Drive = LookupTag(Message,"Drive");
412 ostringstream msg,status;
413 ioprintf(msg,_("Please insert the disc labeled: "
414 "'%s' "
415 "in the drive '%s' and press enter."),
416 Media.c_str(),Drive.c_str());
417 status << "media-change: " // message
418 << Media << ":" // media
419 << Drive << ":" // drive
420 << msg.str() // l10n message
421 << endl;
422 write(status_fd, status.str().c_str(), status.str().size());
423 }
424
425 if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"),
426 LookupTag(Message,"Drive")) == false)
427 {
428 char S[300];
429 snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
430 if (Debug == true)
431 clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
432 OutQueue += S;
433 OutReady = true;
434 return true;
435 }
436
437 char S[300];
438 snprintf(S,sizeof(S),"603 Media Changed\n\n");
439 if (Debug == true)
440 clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
441 OutQueue += S;
442 OutReady = true;
443 return true;
444 }
445 /*}}}*/
446 // Worker::SendConfiguration - Send the config to the method /*{{{*/
447 // ---------------------------------------------------------------------
448 /* */
449 bool pkgAcquire::Worker::SendConfiguration()
450 {
451 if (Config->SendConfig == false)
452 return true;
453
454 if (OutFd == -1)
455 return false;
456
457 string Message = "601 Configuration\n";
458 Message.reserve(2000);
459
460 /* Write out all of the configuration directives by walking the
461 configuration tree */
462 const Configuration::Item *Top = _config->Tree(0);
463 for (; Top != 0;)
464 {
465 if (Top->Value.empty() == false)
466 {
467 string Line = "Config-Item: " + QuoteString(Top->FullTag(),"=\"\n") + "=";
468 Line += QuoteString(Top->Value,"\n") + '\n';
469 Message += Line;
470 }
471
472 if (Top->Child != 0)
473 {
474 Top = Top->Child;
475 continue;
476 }
477
478 while (Top != 0 && Top->Next == 0)
479 Top = Top->Parent;
480 if (Top != 0)
481 Top = Top->Next;
482 }
483 Message += '\n';
484
485 if (Debug == true)
486 clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
487 OutQueue += Message;
488 OutReady = true;
489
490 return true;
491 }
492 /*}}}*/
493 // Worker::QueueItem - Add an item to the outbound queue /*{{{*/
494 // ---------------------------------------------------------------------
495 /* Send a URI Acquire message to the method */
496 bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
497 {
498 if (OutFd == -1)
499 return false;
500
501 string Message = "600 URI Acquire\n";
502 Message.reserve(300);
503 Message += "URI: " + Item->URI;
504 Message += "\nFilename: " + Item->Owner->DestFile;
505 Message += Item->Owner->Custom600Headers();
506 Message += "\n\n";
507
508 if (Debug == true)
509 clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
510 OutQueue += Message;
511 OutReady = true;
512
513 return true;
514 }
515 /*}}}*/
516 // Worker::OutFdRead - Out bound FD is ready /*{{{*/
517 // ---------------------------------------------------------------------
518 /* */
519 bool pkgAcquire::Worker::OutFdReady()
520 {
521 int Res;
522 do
523 {
524 Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
525 }
526 while (Res < 0 && errno == EINTR);
527
528 if (Res <= 0)
529 return MethodFailure();
530
531 OutQueue.erase(0,Res);
532 if (OutQueue.empty() == true)
533 OutReady = false;
534
535 return true;
536 }
537 /*}}}*/
538 // Worker::InFdRead - In bound FD is ready /*{{{*/
539 // ---------------------------------------------------------------------
540 /* */
541 bool pkgAcquire::Worker::InFdReady()
542 {
543 if (ReadMessages() == false)
544 return false;
545 RunMessages();
546 return true;
547 }
548 /*}}}*/
549 // Worker::MethodFailure - Called when the method fails /*{{{*/
550 // ---------------------------------------------------------------------
551 /* This is called when the method is belived to have failed, probably because
552 read returned -1. */
553 bool pkgAcquire::Worker::MethodFailure()
554 {
555 _error->Error("Method %s has died unexpectedly!",Access.c_str());
556
557 // do not reap the child here to show meaningfull error to the user
558 ExecWait(Process,Access.c_str(),false);
559 Process = -1;
560 close(InFd);
561 close(OutFd);
562 InFd = -1;
563 OutFd = -1;
564 OutReady = false;
565 InReady = false;
566 OutQueue = string();
567 MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
568
569 return false;
570 }
571 /*}}}*/
572 // Worker::Pulse - Called periodically /*{{{*/
573 // ---------------------------------------------------------------------
574 /* */
575 void pkgAcquire::Worker::Pulse()
576 {
577 if (CurrentItem == 0)
578 return;
579
580 struct stat Buf;
581 if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
582 return;
583 CurrentSize = Buf.st_size;
584
585 // Hmm? Should not happen...
586 if (CurrentSize > TotalSize && TotalSize != 0)
587 TotalSize = CurrentSize;
588 }
589 /*}}}*/
590 // Worker::ItemDone - Called when the current item is finished /*{{{*/
591 // ---------------------------------------------------------------------
592 /* */
593 void pkgAcquire::Worker::ItemDone()
594 {
595 CurrentItem = 0;
596 CurrentSize = 0;
597 TotalSize = 0;
598 Status = string();
599 }
600 /*}}}*/