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