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