Working acquire code
[ntk/apt.git] / apt-pkg / acquire-worker.cc
CommitLineData
0118833a
AL
1// -*- mode: cpp; mode: fold -*-
2// Description /*{{{*/
bfd22fc0 3// $Id: acquire-worker.cc,v 1.10 1998/11/05 07:21:39 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
25#include <unistd.h>
26#include <signal.h>
93641593 27#include <wait.h>
3b5421b4
AL
28 /*}}}*/
29
30// Worker::Worker - Constructor for Queue startup /*{{{*/
31// ---------------------------------------------------------------------
32/* */
0a8a80e5 33pkgAcquire::Worker::Worker(Queue *Q,MethodConfig *Cnf)
3b5421b4
AL
34{
35 OwnerQ = Q;
0a8a80e5
AL
36 Config = Cnf;
37 Access = Cnf->Access;
38 CurrentItem = 0;
3b5421b4
AL
39
40 Construct();
41}
42 /*}}}*/
43// Worker::Worker - Constructor for method config startup /*{{{*/
44// ---------------------------------------------------------------------
45/* */
46pkgAcquire::Worker::Worker(MethodConfig *Cnf)
47{
48 OwnerQ = 0;
49 Config = Cnf;
50 Access = Cnf->Access;
0a8a80e5
AL
51 CurrentItem = 0;
52
3b5421b4
AL
53 Construct();
54}
55 /*}}}*/
56// Worker::Construct - Constructor helper /*{{{*/
57// ---------------------------------------------------------------------
58/* */
59void pkgAcquire::Worker::Construct()
60{
0a8a80e5
AL
61 NextQueue = 0;
62 NextAcquire = 0;
3b5421b4
AL
63 Process = -1;
64 InFd = -1;
65 OutFd = -1;
0a8a80e5
AL
66 OutReady = false;
67 InReady = false;
3b5421b4
AL
68 Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
69}
70 /*}}}*/
71// Worker::~Worker - Destructor /*{{{*/
72// ---------------------------------------------------------------------
73/* */
74pkgAcquire::Worker::~Worker()
75{
76 close(InFd);
77 close(OutFd);
78
79 if (Process > 0)
0a8a80e5 80 {
3b5421b4 81 kill(Process,SIGINT);
0a8a80e5
AL
82 if (waitpid(Process,0,0) != Process)
83 _error->Warning("I waited but nothing was there!");
84 }
3b5421b4
AL
85}
86 /*}}}*/
87// Worker::Start - Start the worker process /*{{{*/
88// ---------------------------------------------------------------------
89/* This forks the method and inits the communication channel */
90bool pkgAcquire::Worker::Start()
91{
92 // Get the method path
93 string Method = _config->FindDir("Dir::Bin::Methods") + Access;
94 if (FileExists(Method) == false)
95 return _error->Error("The method driver %s could not be found.",Method.c_str());
96
97 if (Debug == true)
98 clog << "Starting method '" << Method << '\'' << endl;
99
100 // Create the pipes
101 int Pipes[4] = {-1,-1,-1,-1};
102 if (pipe(Pipes) != 0 || pipe(Pipes+2) != 0)
103 {
104 _error->Errno("pipe","Failed to create IPC pipe to subprocess");
105 for (int I = 0; I != 4; I++)
106 close(Pipes[I]);
107 return false;
108 }
8b89e57f
AL
109 for (int I = 0; I != 4; I++)
110 SetCloseExec(Pipes[0],true);
111
3b5421b4
AL
112 // Fork off the process
113 Process = fork();
114 if (Process < 0)
115 {
116 cerr << "FATAL -> Failed to fork." << endl;
117 exit(100);
118 }
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;
136 exit(100);
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 }
224 CurrentItem = Itm;
225 CurrentSize = 0;
226 TotalSize = atoi(LookupTag(Message,"Size","0").c_str());
227
228 break;
229 }
0a8a80e5
AL
230
231 // 201 URI Done
232 case 201:
c88edf1d
AL
233 {
234 if (Itm == 0)
235 {
93bf083d 236 _error->Error("Method gave invalid 201 URI Done message");
c88edf1d
AL
237 break;
238 }
239
bfd22fc0 240 pkgAcquire::Item *Owner = Itm->Owner;
be4401bf 241 OwnerQ->ItemDone(Itm);
bfd22fc0 242 Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
c88edf1d 243 LookupTag(Message,"MD5-Hash"));
c88edf1d
AL
244 break;
245 }
0a8a80e5
AL
246
247 // 400 URI Failure
248 case 400:
c88edf1d
AL
249 {
250 if (Itm == 0)
251 {
93bf083d 252 _error->Error("Method gave invalid 400 URI Failure message");
c88edf1d
AL
253 break;
254 }
255
bfd22fc0 256 pkgAcquire::Item *Owner = Itm->Owner;
c88edf1d 257 OwnerQ->ItemDone(Itm);
bfd22fc0 258 Owner->Failed(Message);
c88edf1d
AL
259 break;
260 }
0a8a80e5
AL
261
262 // 401 General Failure
263 case 401:
264 _error->Error("Method %s General failure: %s",LookupTag(Message,"Message").c_str());
265 break;
3b5421b4
AL
266 }
267 }
268 return true;
269}
270 /*}}}*/
271// Worker::Capabilities - 100 Capabilities handler /*{{{*/
272// ---------------------------------------------------------------------
273/* This parses the capabilities message and dumps it into the configuration
274 structure. */
275bool pkgAcquire::Worker::Capabilities(string Message)
276{
277 if (Config == 0)
278 return true;
279
280 Config->Version = LookupTag(Message,"Version");
281 Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
282 Config->PreScan = StringToBool(LookupTag(Message,"Pre-Scan"),false);
0a8a80e5
AL
283 Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
284 Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
3b5421b4
AL
285
286 // Some debug text
287 if (Debug == true)
288 {
289 clog << "Configured access method " << Config->Access << endl;
0a8a80e5
AL
290 clog << "Version:" << Config->Version << " SingleInstance:" <<
291 Config->SingleInstance << " PreScan: " << Config->PreScan <<
292 " Pipeline:" << Config->Pipeline << " SendConfig:" <<
293 Config->SendConfig << endl;
3b5421b4
AL
294 }
295
296 return true;
297}
0118833a 298 /*}}}*/
0a8a80e5
AL
299// Worker::SendConfiguration - Send the config to the method /*{{{*/
300// ---------------------------------------------------------------------
301/* */
302bool pkgAcquire::Worker::SendConfiguration()
303{
304 if (Config->SendConfig == false)
305 return true;
306
307 if (OutFd == -1)
308 return false;
309
310 string Message = "601 Configuration\n";
311 Message.reserve(2000);
312
313 /* Write out all of the configuration directives by walking the
314 configuration tree */
315 const Configuration::Item *Top = _config->Tree(0);
316 for (; Top != 0;)
317 {
318 if (Top->Value.empty() == false)
319 {
320 string Line = "Config-Item: " + Top->FullTag() + "=";
321 Line += QuoteString(Top->Value,"\n") + '\n';
322 Message += Line;
323 }
324
325 if (Top->Child != 0)
326 {
327 Top = Top->Child;
328 continue;
329 }
330
331 while (Top != 0 && Top->Next == 0)
332 Top = Top->Parent;
333 if (Top != 0)
334 Top = Top->Next;
335 }
336 Message += '\n';
337
338 if (Debug == true)
339 clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
340 OutQueue += Message;
341 OutReady = true;
342
343 return true;
344}
345 /*}}}*/
346// Worker::QueueItem - Add an item to the outbound queue /*{{{*/
347// ---------------------------------------------------------------------
348/* Send a URI Acquire message to the method */
349bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
350{
351 if (OutFd == -1)
352 return false;
353
354 string Message = "600 URI Acquire\n";
355 Message.reserve(300);
356 Message += "URI: " + Item->URI;
357 Message += "\nFilename: " + Item->Owner->DestFile;
358 Message += Item->Owner->Custom600Headers();
359 Message += "\n\n";
360
361 if (Debug == true)
362 clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
363 OutQueue += Message;
364 OutReady = true;
365
366 return true;
367}
368 /*}}}*/
369// Worker::OutFdRead - Out bound FD is ready /*{{{*/
370// ---------------------------------------------------------------------
371/* */
372bool pkgAcquire::Worker::OutFdReady()
373{
374 int Res = write(OutFd,OutQueue.begin(),OutQueue.length());
375 if (Res <= 0)
376 return MethodFailure();
377
378 // Hmm.. this should never happen.
379 if (Res < 0)
380 return true;
381
382 OutQueue.erase(0,Res);
383 if (OutQueue.empty() == true)
384 OutReady = false;
385
386 return true;
387}
388 /*}}}*/
389// Worker::InFdRead - In bound FD is ready /*{{{*/
390// ---------------------------------------------------------------------
391/* */
392bool pkgAcquire::Worker::InFdReady()
393{
394 if (ReadMessages() == false)
395 return false;
396 RunMessages();
397 return true;
398}
399 /*}}}*/
400// Worker::MethodFailure - Called when the method fails /*{{{*/
401// ---------------------------------------------------------------------
402/* This is called when the method is belived to have failed, probably because
403 read returned -1. */
404bool pkgAcquire::Worker::MethodFailure()
405{
406 cerr << "Method " << Access << " has died unexpectedly!" << endl;
407 if (waitpid(Process,0,0) != Process)
408 _error->Warning("I waited but nothing was there!");
409 Process = -1;
410 close(InFd);
411 close(OutFd);
412 InFd = -1;
413 OutFd = -1;
414 OutReady = false;
415 InReady = false;
416 OutQueue = string();
417 MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
418
419 return false;
420}
421 /*}}}*/