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