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