apt-pkg/acquire-worker.cc

   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 <fstream>
  29
  30 #include <sys/stat.h>
  31 #include <unistd.h>
  32 #include <fcntl.h>
  33 #include <signal.h>
  34 #include <stdio.h>
  35 #include <errno.h>
  36                                                                         /*}}}*/
  37
  38 using namespace std;
  39
  40 // Worker::Worker - Constructor for Queue startup                       /*{{{*/
  41 // ---------------------------------------------------------------------
  42 /* */
  43 pkgAcquire::Worker::Worker(Queue *Q,MethodConfig *Cnf,
  44                            pkgAcquireStatus *Log) : Log(Log)
  45 {
  46    OwnerQ = Q;
  47    Config = Cnf;
  48    Access = Cnf->Access;
  49    CurrentItem = 0;
  50    TotalSize = 0;
  51    CurrentSize = 0;
  52
  53    Construct();
  54 }
  55                                                                         /*}}}*/
  56 // Worker::Worker - Constructor for method config startup               /*{{{*/
  57 // ---------------------------------------------------------------------
  58 /* */
  59 pkgAcquire::Worker::Worker(MethodConfig *Cnf)
  60 {
  61    OwnerQ = 0;
  62    Config = Cnf;
  63    Access = Cnf->Access;
  64    CurrentItem = 0;
  65    TotalSize = 0;
  66    CurrentSize = 0;
  67
  68    Construct();
  69 }
  70                                                                         /*}}}*/
  71 // Worker::Construct - Constructor helper                               /*{{{*/
  72 // ---------------------------------------------------------------------
  73 /* */
  74 void pkgAcquire::Worker::Construct()
  75 {
  76    NextQueue = 0;
  77    NextAcquire = 0;
  78    Process = -1;
  79    InFd = -1;
  80    OutFd = -1;
  81    OutReady = false;
  82    InReady = false;
  83    Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
  84 }
  85                                                                         /*}}}*/
  86 // Worker::~Worker - Destructor                                         /*{{{*/
  87 // ---------------------------------------------------------------------
  88 /* */
  89 pkgAcquire::Worker::~Worker()
  90 {
  91    close(InFd);
  92    close(OutFd);
  93
  94    if (Process > 0)
  95    {
  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);
 100       ExecWait(Process,Access.c_str(),true);
 101    }
 102 }
 103                                                                         /*}}}*/
 104 // Worker::Start - Start the worker process                             /*{{{*/
 105 // ---------------------------------------------------------------------
 106 /* This forks the method and inits the communication channel */
 107 bool pkgAcquire::Worker::Start()
 108 {
 109    // Get the method path
 110    string Method = _config->FindDir("Dir::Bin::Methods") + Access;
 111    if (FileExists(Method) == false)
 112       return _error->Error(_("The method driver %s could not be found."),Method.c_str());
 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    }
 126    for (int I = 0; I != 4; I++)
 127       SetCloseExec(Pipes[I],true);
 128
 129    // Fork off the process
 130    Process = ExecFork();
 131    if (Process == 0)
 132    {
 133       // Setup the FDs
 134       dup2(Pipes[1],STDOUT_FILENO);
 135       dup2(Pipes[2],STDIN_FILENO);
 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;
 145       _exit(100);
 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]);
 155    OutReady = false;
 156    InReady = true;
 157
 158    // Read the configuration data
 159    if (WaitFd(InFd) == false ||
 160        ReadMessages() == false)
 161       return _error->Error(_("Method %s did not start correctly"),Method.c_str());
 162
 163    RunMessages();
 164    if (OwnerQ != 0)
 165       SendConfiguration();
 166
 167    return true;
 168 }
 169                                                                         /*}}}*/
 170 // Worker::ReadMessages - Read all pending messages into the list       /*{{{*/
 171 // ---------------------------------------------------------------------
 172 /* */
 173 bool pkgAcquire::Worker::ReadMessages()
 174 {
 175    if (::ReadMessages(InFd,MessageQueue) == false)
 176       return MethodFailure();
 177    return true;
 178 }
 179                                                                         /*}}}*/
 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. */
 184 bool pkgAcquire::Worker::RunMessages()
 185 {
 186    while (MessageQueue.empty() == false)
 187    {
 188       string Message = MessageQueue.front();
 189       MessageQueue.erase(MessageQueue.begin());
 190
 191       if (Debug == true)
 192          clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
 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
 200       string URI = LookupTag(Message,"URI");
 201       pkgAcquire::Queue::QItem *Itm = 0;
 202       if (URI.empty() == false)
 203          Itm = OwnerQ->FindItem(URI,this);
 204
 205       // Determine the message number and dispatch
 206       switch (Number)
 207       {
 208          // 100 Capabilities
 209          case 100:
 210          if (Capabilities(Message) == false)
 211             return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
 212          break;
 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:
 227          {
 228             if (Itm == 0)
 229             {
 230                _error->Error("Method gave invalid 200 URI Start message");
 231                break;
 232             }
 233
 234             CurrentItem = Itm;
 235             CurrentSize = 0;
 236             TotalSize = atoi(LookupTag(Message,"Size","0").c_str());
 237             ResumePoint = atoi(LookupTag(Message,"Resume-Point","0").c_str());
 238             Itm->Owner->Start(Message,atoi(LookupTag(Message,"Size","0").c_str()));
 239
 240             // Display update before completion
 241             if (Log != 0 && Log->MorePulses == true)
 242                Log->Pulse(Itm->Owner->GetOwner());
 243
 244             if (Log != 0)
 245                Log->Fetch(*Itm);
 246
 247             break;
 248          }
 249
 250          // 201 URI Done
 251          case 201:
 252          {
 253             if (Itm == 0)
 254             {
 255                _error->Error("Method gave invalid 201 URI Done message");
 256                break;
 257             }
 258
 259             pkgAcquire::Item *Owner = Itm->Owner;
 260             pkgAcquire::ItemDesc Desc = *Itm;
 261
 262             // Display update before completion
 263             if (Log != 0 && Log->MorePulses == true)
 264                Log->Pulse(Owner->GetOwner());
 265
 266             OwnerQ->ItemDone(Itm);
 267             if (TotalSize != 0 &&
 268                 (unsigned)atoi(LookupTag(Message,"Size","0").c_str()) != TotalSize)
 269                _error->Warning("Bizarre Error - File size is not what the server reported %s %lu",
 270                                LookupTag(Message,"Size","0").c_str(),TotalSize);
 271
 272             Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
 273                         LookupTag(Message,"MD5-Hash"),Config);
 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)
 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                }
 287                else
 288                   Log->Done(Desc);
 289             }
 290             break;
 291          }
 292
 293          // 400 URI Failure
 294          case 400:
 295          {
 296             if (Itm == 0)
 297             {
 298                _error->Error("Method gave invalid 400 URI Failure message");
 299                break;
 300             }
 301
 302             // Display update before completion
 303             if (Log != 0 && Log->MorePulses == true)
 304                Log->Pulse(Itm->Owner->GetOwner());
 305
 306             pkgAcquire::Item *Owner = Itm->Owner;
 307             pkgAcquire::ItemDesc Desc = *Itm;
 308             OwnerQ->ItemDone(Itm);
 309             Owner->Failed(Message,Config);
 310             ItemDone();
 311
 312             if (Log != 0)
 313                Log->Fail(Desc);
 314
 315             break;
 316          }
 317
 318          // 401 General Failure
 319          case 401:
 320          _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
 321          break;
 322
 323          // 403 Media Change
 324          case 403:
 325          MediaChange(Message);
 326          break;
 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. */
 336 bool 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);
 343    Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
 344    Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
 345    Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
 346    Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
 347    Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);
 348
 349    // Some debug text
 350    if (Debug == true)
 351    {
 352       clog << "Configured access method " << Config->Access << endl;
 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;
 360    }
 361
 362    return true;
 363 }
 364                                                                         /*}}}*/
 365 // Worker::MediaChange - Request a media change                         /*{{{*/
 366 // ---------------------------------------------------------------------
 367 /* */
 368 bool 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];
 374       snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
 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];
 383    snprintf(S,sizeof(S),"603 Media Changed\n\n");
 384    if (Debug == true)
 385       clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
 386    OutQueue += S;
 387    OutReady = true;
 388    return true;
 389 }
 390                                                                         /*}}}*/
 391 // Worker::SendConfiguration - Send the config to the method            /*{{{*/
 392 // ---------------------------------------------------------------------
 393 /* */
 394 bool 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       {
 412          string Line = "Config-Item: " + QuoteString(Top->FullTag(),"=\"\n") + "=";
 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 */
 441 bool 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 /* */
 464 bool pkgAcquire::Worker::OutFdReady()
 465 {
 466    int Res;
 467    do
 468    {
 469       Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
 470    }
 471    while (Res < 0 && errno == EINTR);
 472
 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 /* */
 490 bool 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. */
 502 bool pkgAcquire::Worker::MethodFailure()
 503 {
 504    _error->Error("Method %s has died unexpectedly!",Access.c_str());
 505
 506    ExecWait(Process,Access.c_str(),true);
 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                                                                         /*}}}*/
 520 // Worker::Pulse - Called periodically                                  /*{{{*/
 521 // ---------------------------------------------------------------------
 522 /* */
 523 void pkgAcquire::Worker::Pulse()
 524 {
 525    if (CurrentItem == 0)
 526       return;
 527
 528    struct stat Buf;
 529    if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
 530       return;
 531    CurrentSize = Buf.st_size;
 532
 533    // Hmm? Should not happen...
 534    if (CurrentSize > TotalSize && TotalSize != 0)
 535       TotalSize = CurrentSize;
 536 }
 537                                                                         /*}}}*/
 538 // Worker::ItemDone - Called when the current item is finished          /*{{{*/
 539 // ---------------------------------------------------------------------
 540 /* */
 541 void pkgAcquire::Worker::ItemDone()
 542 {
 543    CurrentItem = 0;
 544    CurrentSize = 0;
 545    TotalSize = 0;
 546    Status = string();
 547 }
 548                                                                         /*}}}*/