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