apt-pkg/acquire-worker.cc

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