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
 311             // set some status
 312             if(LookupTag(Message,"FailReason") == "Timeout" ||
 313                LookupTag(Message,"FailReason") == "TmpResolveFailure" ||
 314                LookupTag(Message,"FailReason") == "ConnectionRefused")
 315                Owner->Status = pkgAcquire::Item::StatTransientNetworkError;
 316
 317             Owner->Failed(Message,Config);
 318             ItemDone();
 319
 320             if (Log != 0)
 321                Log->Fail(Desc);
 322
 323             break;
 324          }
 325
 326          // 401 General Failure
 327          case 401:
 328          _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
 329          break;
 330
 331          // 403 Media Change
 332          case 403:
 333          MediaChange(Message);
 334          break;
 335       }
 336    }
 337    return true;
 338 }
 339                                                                         /*}}}*/
 340 // Worker::Capabilities - 100 Capabilities handler                      /*{{{*/
 341 // ---------------------------------------------------------------------
 342 /* This parses the capabilities message and dumps it into the configuration
 343    structure. */
 344 bool pkgAcquire::Worker::Capabilities(string Message)
 345 {
 346    if (Config == 0)
 347       return true;
 348
 349    Config->Version = LookupTag(Message,"Version");
 350    Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
 351    Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
 352    Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
 353    Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
 354    Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
 355    Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);
 356
 357    // Some debug text
 358    if (Debug == true)
 359    {
 360       clog << "Configured access method " << Config->Access << endl;
 361       clog << "Version:" << Config->Version <<
 362               " SingleInstance:" << Config->SingleInstance <<
 363               " Pipeline:" << Config->Pipeline <<
 364               " SendConfig:" << Config->SendConfig <<
 365               " LocalOnly: " << Config->LocalOnly <<
 366               " NeedsCleanup: " << Config->NeedsCleanup <<
 367               " Removable: " << Config->Removable << endl;
 368    }
 369
 370    return true;
 371 }
 372                                                                         /*}}}*/
 373 // Worker::MediaChange - Request a media change                         /*{{{*/
 374 // ---------------------------------------------------------------------
 375 /* */
 376 bool pkgAcquire::Worker::MediaChange(string Message)
 377 {
 378    int status_fd = _config->FindI("APT::Status-Fd",-1);
 379    if(status_fd > 0)
 380    {
 381       string Media = LookupTag(Message,"Media");
 382       string Drive = LookupTag(Message,"Drive");
 383       ostringstream msg,status;
 384       ioprintf(msg,_("Please insert the disc labeled: "
 385                      "'%s' "
 386                      "in the drive '%s' and press enter."),
 387                Media.c_str(),Drive.c_str());
 388       status << "media-change: "  // message
 389              << Media  << ":"     // media
 390              << Drive  << ":"     // drive
 391              << msg.str()         // l10n message
 392              << endl;
 393       write(status_fd, status.str().c_str(), status.str().size());
 394    }
 395
 396    if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"),
 397                                     LookupTag(Message,"Drive")) == false)
 398    {
 399       char S[300];
 400       snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
 401       if (Debug == true)
 402          clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
 403       OutQueue += S;
 404       OutReady = true;
 405       return true;
 406    }
 407
 408    char S[300];
 409    snprintf(S,sizeof(S),"603 Media Changed\n\n");
 410    if (Debug == true)
 411       clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
 412    OutQueue += S;
 413    OutReady = true;
 414    return true;
 415 }
 416                                                                         /*}}}*/
 417 // Worker::SendConfiguration - Send the config to the method            /*{{{*/
 418 // ---------------------------------------------------------------------
 419 /* */
 420 bool pkgAcquire::Worker::SendConfiguration()
 421 {
 422    if (Config->SendConfig == false)
 423       return true;
 424
 425    if (OutFd == -1)
 426       return false;
 427
 428    string Message = "601 Configuration\n";
 429    Message.reserve(2000);
 430
 431    /* Write out all of the configuration directives by walking the
 432       configuration tree */
 433    const Configuration::Item *Top = _config->Tree(0);
 434    for (; Top != 0;)
 435    {
 436       if (Top->Value.empty() == false)
 437       {
 438          string Line = "Config-Item: " + QuoteString(Top->FullTag(),"=\"\n") + "=";
 439          Line += QuoteString(Top->Value,"\n") + '\n';
 440          Message += Line;
 441       }
 442
 443       if (Top->Child != 0)
 444       {
 445          Top = Top->Child;
 446          continue;
 447       }
 448
 449       while (Top != 0 && Top->Next == 0)
 450          Top = Top->Parent;
 451       if (Top != 0)
 452          Top = Top->Next;
 453    }
 454    Message += '\n';
 455
 456    if (Debug == true)
 457       clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
 458    OutQueue += Message;
 459    OutReady = true;
 460
 461    return true;
 462 }
 463                                                                         /*}}}*/
 464 // Worker::QueueItem - Add an item to the outbound queue                /*{{{*/
 465 // ---------------------------------------------------------------------
 466 /* Send a URI Acquire message to the method */
 467 bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
 468 {
 469    if (OutFd == -1)
 470       return false;
 471
 472    string Message = "600 URI Acquire\n";
 473    Message.reserve(300);
 474    Message += "URI: " + Item->URI;
 475    Message += "\nFilename: " + Item->Owner->DestFile;
 476    Message += Item->Owner->Custom600Headers();
 477    Message += "\n\n";
 478
 479    if (Debug == true)
 480       clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
 481    OutQueue += Message;
 482    OutReady = true;
 483
 484    return true;
 485 }
 486                                                                         /*}}}*/
 487 // Worker::OutFdRead - Out bound FD is ready                            /*{{{*/
 488 // ---------------------------------------------------------------------
 489 /* */
 490 bool pkgAcquire::Worker::OutFdReady()
 491 {
 492    int Res;
 493    do
 494    {
 495       Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
 496    }
 497    while (Res < 0 && errno == EINTR);
 498
 499    if (Res <= 0)
 500       return MethodFailure();
 501
 502    // Hmm.. this should never happen.
 503    if (Res < 0)
 504       return true;
 505
 506    OutQueue.erase(0,Res);
 507    if (OutQueue.empty() == true)
 508       OutReady = false;
 509
 510    return true;
 511 }
 512                                                                         /*}}}*/
 513 // Worker::InFdRead - In bound FD is ready                              /*{{{*/
 514 // ---------------------------------------------------------------------
 515 /* */
 516 bool pkgAcquire::Worker::InFdReady()
 517 {
 518    if (ReadMessages() == false)
 519       return false;
 520    RunMessages();
 521    return true;
 522 }
 523                                                                         /*}}}*/
 524 // Worker::MethodFailure - Called when the method fails                 /*{{{*/
 525 // ---------------------------------------------------------------------
 526 /* This is called when the method is belived to have failed, probably because
 527    read returned -1. */
 528 bool pkgAcquire::Worker::MethodFailure()
 529 {
 530    _error->Error("Method %s has died unexpectedly!",Access.c_str());
 531
 532    ExecWait(Process,Access.c_str(),true);
 533    Process = -1;
 534    close(InFd);
 535    close(OutFd);
 536    InFd = -1;
 537    OutFd = -1;
 538    OutReady = false;
 539    InReady = false;
 540    OutQueue = string();
 541    MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
 542
 543    return false;
 544 }
 545                                                                         /*}}}*/
 546 // Worker::Pulse - Called periodically                                  /*{{{*/
 547 // ---------------------------------------------------------------------
 548 /* */
 549 void pkgAcquire::Worker::Pulse()
 550 {
 551    if (CurrentItem == 0)
 552       return;
 553
 554    struct stat Buf;
 555    if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
 556       return;
 557    CurrentSize = Buf.st_size;
 558
 559    // Hmm? Should not happen...
 560    if (CurrentSize > TotalSize && TotalSize != 0)
 561       TotalSize = CurrentSize;
 562 }
 563                                                                         /*}}}*/
 564 // Worker::ItemDone - Called when the current item is finished          /*{{{*/
 565 // ---------------------------------------------------------------------
 566 /* */
 567 void pkgAcquire::Worker::ItemDone()
 568 {
 569    CurrentItem = 0;
 570    CurrentSize = 0;
 571    TotalSize = 0;
 572    Status = string();
 573 }
 574                                                                         /*}}}*/