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