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 #include <config.h>
  16
  17 #include <apt-pkg/acquire.h>
  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 #include <apt-pkg/hashes.h>
  25
  26 #include <string>
  27 #include <vector>
  28 #include <iostream>
  29
  30 #include <sys/stat.h>
  31 #include <stdlib.h>
  32 #include <unistd.h>
  33 #include <signal.h>
  34 #include <stdio.h>
  35 #include <errno.h>
  36 #include <sstream>
  37
  38 #include <apti18n.h>
  39                                                                         /*}}}*/
  40
  41 using namespace std;
  42
  43 // Worker::Worker - Constructor for Queue startup                       /*{{{*/
  44 // ---------------------------------------------------------------------
  45 /* */
  46 pkgAcquire::Worker::Worker(Queue *Q,MethodConfig *Cnf,
  47                            pkgAcquireStatus *log) : d(NULL), Log(log)
  48 {
  49    OwnerQ = Q;
  50    Config = Cnf;
  51    Access = Cnf->Access;
  52    CurrentItem = 0;
  53    TotalSize = 0;
  54    CurrentSize = 0;
  55
  56    Construct();
  57 }
  58                                                                         /*}}}*/
  59 // Worker::Worker - Constructor for method config startup               /*{{{*/
  60 // ---------------------------------------------------------------------
  61 /* */
  62 pkgAcquire::Worker::Worker(MethodConfig *Cnf) : d(NULL), OwnerQ(NULL), Config(Cnf),
  63                                                 Access(Cnf->Access), CurrentItem(NULL),
  64                                                 CurrentSize(0), TotalSize(0)
  65 {
  66    Construct();
  67 }
  68                                                                         /*}}}*/
  69 // Worker::Construct - Constructor helper                               /*{{{*/
  70 // ---------------------------------------------------------------------
  71 /* */
  72 void pkgAcquire::Worker::Construct()
  73 {
  74    NextQueue = 0;
  75    NextAcquire = 0;
  76    Process = -1;
  77    InFd = -1;
  78    OutFd = -1;
  79    OutReady = false;
  80    InReady = false;
  81    Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
  82 }
  83                                                                         /*}}}*/
  84 // Worker::~Worker - Destructor                                         /*{{{*/
  85 // ---------------------------------------------------------------------
  86 /* */
  87 pkgAcquire::Worker::~Worker()
  88 {
  89    close(InFd);
  90    close(OutFd);
  91
  92    if (Process > 0)
  93    {
  94       /* Closing of stdin is the signal to exit and die when the process
  95          indicates it needs cleanup */
  96       if (Config->NeedsCleanup == false)
  97          kill(Process,SIGINT);
  98       ExecWait(Process,Access.c_str(),true);
  99    }
 100 }
 101                                                                         /*}}}*/
 102 // Worker::Start - Start the worker process                             /*{{{*/
 103 // ---------------------------------------------------------------------
 104 /* This forks the method and inits the communication channel */
 105 bool pkgAcquire::Worker::Start()
 106 {
 107    // Get the method path
 108    string Method = _config->FindDir("Dir::Bin::Methods") + Access;
 109    if (FileExists(Method) == false)
 110    {
 111       _error->Error(_("The method driver %s could not be found."),Method.c_str());
 112       if (Access == "https")
 113          _error->Notice(_("Is the package %s installed?"), "apt-transport-https");
 114       return false;
 115    }
 116
 117    if (Debug == true)
 118       clog << "Starting method '" << Method << '\'' << endl;
 119
 120    // Create the pipes
 121    int Pipes[4] = {-1,-1,-1,-1};
 122    if (pipe(Pipes) != 0 || pipe(Pipes+2) != 0)
 123    {
 124       _error->Errno("pipe","Failed to create IPC pipe to subprocess");
 125       for (int I = 0; I != 4; I++)
 126          close(Pipes[I]);
 127       return false;
 128    }
 129    for (int I = 0; I != 4; I++)
 130       SetCloseExec(Pipes[I],true);
 131
 132    // Fork off the process
 133    Process = ExecFork();
 134    if (Process == 0)
 135    {
 136       // Setup the FDs
 137       dup2(Pipes[1],STDOUT_FILENO);
 138       dup2(Pipes[2],STDIN_FILENO);
 139       SetCloseExec(STDOUT_FILENO,false);
 140       SetCloseExec(STDIN_FILENO,false);
 141       SetCloseExec(STDERR_FILENO,false);
 142
 143       const char *Args[2];
 144       Args[0] = Method.c_str();
 145       Args[1] = 0;
 146       execv(Args[0],(char **)Args);
 147       cerr << "Failed to exec method " << Args[0] << endl;
 148       _exit(100);
 149    }
 150
 151    // Fix up our FDs
 152    InFd = Pipes[0];
 153    OutFd = Pipes[3];
 154    SetNonBlock(Pipes[0],true);
 155    SetNonBlock(Pipes[3],true);
 156    close(Pipes[1]);
 157    close(Pipes[2]);
 158    OutReady = false;
 159    InReady = true;
 160
 161    // Read the configuration data
 162    if (WaitFd(InFd) == false ||
 163        ReadMessages() == false)
 164       return _error->Error(_("Method %s did not start correctly"),Method.c_str());
 165
 166    RunMessages();
 167    if (OwnerQ != 0)
 168       SendConfiguration();
 169
 170    return true;
 171 }
 172                                                                         /*}}}*/
 173 // Worker::ReadMessages - Read all pending messages into the list       /*{{{*/
 174 // ---------------------------------------------------------------------
 175 /* */
 176 bool pkgAcquire::Worker::ReadMessages()
 177 {
 178    if (::ReadMessages(InFd,MessageQueue) == false)
 179       return MethodFailure();
 180    return true;
 181 }
 182                                                                         /*}}}*/
 183 // Worker::RunMessage - Empty the message queue                         /*{{{*/
 184 // ---------------------------------------------------------------------
 185 /* This takes the messages from the message queue and runs them through
 186    the parsers in order. */
 187 bool pkgAcquire::Worker::RunMessages()
 188 {
 189    while (MessageQueue.empty() == false)
 190    {
 191       string Message = MessageQueue.front();
 192       MessageQueue.erase(MessageQueue.begin());
 193
 194       if (Debug == true)
 195          clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
 196
 197       // Fetch the message number
 198       char *End;
 199       int Number = strtol(Message.c_str(),&End,10);
 200       if (End == Message.c_str())
 201          return _error->Error("Invalid message from method %s: %s",Access.c_str(),Message.c_str());
 202
 203       string URI = LookupTag(Message,"URI");
 204       pkgAcquire::Queue::QItem *Itm = NULL;
 205       if (URI.empty() == false)
 206          Itm = OwnerQ->FindItem(URI,this);
 207
 208       if (Itm != NULL)
 209       {
 210          // update used mirror
 211          string UsedMirror = LookupTag(Message,"UsedMirror", "");
 212          if (UsedMirror.empty() == false)
 213          {
 214             for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 215                (*O)->UsedMirror = UsedMirror;
 216
 217             if (Itm->Description.find(" ") != string::npos)
 218                Itm->Description.replace(0, Itm->Description.find(" "), UsedMirror);
 219          }
 220       }
 221
 222       // Determine the message number and dispatch
 223       switch (Number)
 224       {
 225          // 100 Capabilities
 226          case 100:
 227          if (Capabilities(Message) == false)
 228             return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
 229          break;
 230
 231          // 101 Log
 232          case 101:
 233          if (Debug == true)
 234             clog << " <- (log) " << LookupTag(Message,"Message") << endl;
 235          break;
 236
 237          // 102 Status
 238          case 102:
 239          Status = LookupTag(Message,"Message");
 240          break;
 241
 242          // 103 Redirect
 243          case 103:
 244          {
 245             if (Itm == 0)
 246             {
 247                _error->Error("Method gave invalid 103 Redirect message");
 248                break;
 249             }
 250
 251             std::string const NewURI = LookupTag(Message,"New-URI",URI.c_str());
 252             Itm->URI = NewURI;
 253
 254             ItemDone();
 255
 256             // Change the status so that it can be dequeued
 257             for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 258                (*O)->Status = pkgAcquire::Item::StatIdle;
 259             // Mark the item as done (taking care of all queues)
 260             // and then put it in the main queue again
 261             std::vector<Item*> const ItmOwners = Itm->Owners;
 262             OwnerQ->ItemDone(Itm);
 263             Itm = NULL;
 264             for (pkgAcquire::Queue::QItem::owner_iterator O = ItmOwners.begin(); O != ItmOwners.end(); ++O)
 265             {
 266                pkgAcquire::Item *Owner = *O;
 267                pkgAcquire::ItemDesc &desc = Owner->GetItemDesc();
 268                // if we change site, treat it as a mirror change
 269                if (URI::SiteOnly(NewURI) != URI::SiteOnly(desc.URI))
 270                {
 271                   std::string const OldSite = desc.Description.substr(0, desc.Description.find(" "));
 272                   if (likely(APT::String::Startswith(desc.URI, OldSite)))
 273                   {
 274                      std::string const OldExtra = desc.URI.substr(OldSite.length() + 1);
 275                      if (likely(APT::String::Endswith(NewURI, OldExtra)))
 276                      {
 277                         std::string const NewSite = NewURI.substr(0, NewURI.length() - OldExtra.length());
 278                         Owner->UsedMirror = URI::ArchiveOnly(NewSite);
 279                         if (desc.Description.find(" ") != string::npos)
 280                            desc.Description.replace(0, desc.Description.find(" "), Owner->UsedMirror);
 281                      }
 282                   }
 283                }
 284                desc.URI = NewURI;
 285                OwnerQ->Owner->Enqueue(desc);
 286
 287                if (Log != 0)
 288                   Log->Done(desc);
 289             }
 290             break;
 291          }
 292
 293          // 200 URI Start
 294          case 200:
 295          {
 296             if (Itm == 0)
 297             {
 298                _error->Error("Method gave invalid 200 URI Start message");
 299                break;
 300             }
 301
 302             CurrentItem = Itm;
 303             CurrentSize = 0;
 304             TotalSize = strtoull(LookupTag(Message,"Size","0").c_str(), NULL, 10);
 305             ResumePoint = strtoull(LookupTag(Message,"Resume-Point","0").c_str(), NULL, 10);
 306             for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 307             {
 308                (*O)->Start(Message, TotalSize);
 309
 310                // Display update before completion
 311                if (Log != 0)
 312                {
 313                   if (Log->MorePulses == true)
 314                      Log->Pulse((*O)->GetOwner());
 315                   Log->Fetch((*O)->GetItemDesc());
 316                }
 317             }
 318
 319             break;
 320          }
 321
 322          // 201 URI Done
 323          case 201:
 324          {
 325             if (Itm == 0)
 326             {
 327                _error->Error("Method gave invalid 201 URI Done message");
 328                break;
 329             }
 330
 331             PrepareFiles("201::URIDone", Itm);
 332
 333             // Display update before completion
 334             if (Log != 0 && Log->MorePulses == true)
 335                for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 336                   Log->Pulse((*O)->GetOwner());
 337
 338             std::string const filename = LookupTag(Message, "Filename", Itm->Owner->DestFile.c_str());
 339             HashStringList ReceivedHashes;
 340             {
 341                // see if we got hashes to verify
 342                for (char const * const * type = HashString::SupportedHashes(); *type != NULL; ++type)
 343                {
 344                   std::string const tagname = std::string(*type) + "-Hash";
 345                   std::string const hashsum = LookupTag(Message, tagname.c_str());
 346                   if (hashsum.empty() == false)
 347                      ReceivedHashes.push_back(HashString(*type, hashsum));
 348                }
 349                // not all methods always sent Hashes our way
 350                if (ReceivedHashes.usable() == false)
 351                {
 352                   HashStringList const ExpectedHashes = Itm->GetExpectedHashes();
 353                   if (ExpectedHashes.usable() == true && RealFileExists(filename))
 354                   {
 355                      Hashes calc(ExpectedHashes);
 356                      FileFd file(filename, FileFd::ReadOnly, FileFd::None);
 357                      calc.AddFD(file);
 358                      ReceivedHashes = calc.GetHashStringList();
 359                   }
 360                }
 361             }
 362
 363             // only local files can refer other filenames and counting them as fetched would be unfair
 364             if (Log !=  NULL && filename != Itm->Owner->DestFile)
 365                Log->Fetched(ReceivedHashes.FileSize(),atoi(LookupTag(Message,"Resume-Point","0").c_str()));
 366
 367             std::vector<Item*> const ItmOwners = Itm->Owners;
 368             OwnerQ->ItemDone(Itm);
 369             Itm = NULL;
 370
 371             bool const isIMSHit = StringToBool(LookupTag(Message,"IMS-Hit"),false) ||
 372                StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false);
 373             for (pkgAcquire::Queue::QItem::owner_iterator O = ItmOwners.begin(); O != ItmOwners.end(); ++O)
 374             {
 375                pkgAcquire::Item * const Owner = *O;
 376                HashStringList const ExpectedHashes = Owner->GetExpectedHashes();
 377                if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
 378                {
 379                   std::clog << "201 URI Done: " << Owner->DescURI() << endl
 380                      << "ReceivedHash:" << endl;
 381                   for (HashStringList::const_iterator hs = ReceivedHashes.begin(); hs != ReceivedHashes.end(); ++hs)
 382                      std::clog <<  "\t- " << hs->toStr() << std::endl;
 383                   std::clog << "ExpectedHash:" << endl;
 384                   for (HashStringList::const_iterator hs = ExpectedHashes.begin(); hs != ExpectedHashes.end(); ++hs)
 385                      std::clog <<  "\t- " << hs->toStr() << std::endl;
 386                   std::clog << endl;
 387                }
 388
 389                // decide if what we got is what we expected
 390                bool consideredOkay = false;
 391                if (ExpectedHashes.usable())
 392                {
 393                   if (ReceivedHashes.usable() == false)
 394                   {
 395                      /* IMS-Hits can't be checked here as we will have uncompressed file,
 396                         but the hashes for the compressed file. What we have was good through
 397                         so all we have to ensure later is that we are not stalled. */
 398                      consideredOkay = isIMSHit;
 399                   }
 400                   else if (ReceivedHashes == ExpectedHashes)
 401                      consideredOkay = true;
 402                   else
 403                      consideredOkay = false;
 404
 405                }
 406                else if (Owner->HashesRequired() == true)
 407                   consideredOkay = false;
 408                else
 409                   consideredOkay = true;
 410
 411                if (consideredOkay == true)
 412                   consideredOkay = Owner->VerifyDone(Message, Config);
 413                else // hashsum mismatch
 414                   Owner->Status = pkgAcquire::Item::StatAuthError;
 415
 416                if (consideredOkay == true)
 417                {
 418                   Owner->Done(Message, ReceivedHashes, Config);
 419                   if (Log != 0)
 420                   {
 421                      if (isIMSHit)
 422                         Log->IMSHit(Owner->GetItemDesc());
 423                      else
 424                         Log->Done(Owner->GetItemDesc());
 425                   }
 426                }
 427                else
 428                {
 429                   Owner->Failed(Message,Config);
 430                   if (Log != 0)
 431                      Log->Fail(Owner->GetItemDesc());
 432                }
 433             }
 434             ItemDone();
 435             break;
 436          }
 437
 438          // 400 URI Failure
 439          case 400:
 440          {
 441             if (Itm == 0)
 442             {
 443                std::string const msg = LookupTag(Message,"Message");
 444                _error->Error("Method gave invalid 400 URI Failure message: %s", msg.c_str());
 445                break;
 446             }
 447
 448             PrepareFiles("400::URIFailure", Itm);
 449
 450             // Display update before completion
 451             if (Log != 0 && Log->MorePulses == true)
 452                for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 453                   Log->Pulse((*O)->GetOwner());
 454
 455             std::vector<Item*> const ItmOwners = Itm->Owners;
 456             OwnerQ->ItemDone(Itm);
 457             Itm = NULL;
 458
 459             for (pkgAcquire::Queue::QItem::owner_iterator O = ItmOwners.begin(); O != ItmOwners.end(); ++O)
 460             {
 461                // set some status
 462                if(LookupTag(Message,"FailReason") == "Timeout" ||
 463                      LookupTag(Message,"FailReason") == "TmpResolveFailure" ||
 464                      LookupTag(Message,"FailReason") == "ResolveFailure" ||
 465                      LookupTag(Message,"FailReason") == "ConnectionRefused")
 466                   (*O)->Status = pkgAcquire::Item::StatTransientNetworkError;
 467
 468                (*O)->Failed(Message,Config);
 469
 470                if (Log != 0)
 471                   Log->Fail((*O)->GetItemDesc());
 472             }
 473             ItemDone();
 474
 475             break;
 476          }
 477
 478          // 401 General Failure
 479          case 401:
 480          _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
 481          break;
 482
 483          // 403 Media Change
 484          case 403:
 485          MediaChange(Message);
 486          break;
 487       }
 488    }
 489    return true;
 490 }
 491                                                                         /*}}}*/
 492 // Worker::Capabilities - 100 Capabilities handler                      /*{{{*/
 493 // ---------------------------------------------------------------------
 494 /* This parses the capabilities message and dumps it into the configuration
 495    structure. */
 496 bool pkgAcquire::Worker::Capabilities(string Message)
 497 {
 498    if (Config == 0)
 499       return true;
 500
 501    Config->Version = LookupTag(Message,"Version");
 502    Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
 503    Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
 504    Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
 505    Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
 506    Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
 507    Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);
 508
 509    // Some debug text
 510    if (Debug == true)
 511    {
 512       clog << "Configured access method " << Config->Access << endl;
 513       clog << "Version:" << Config->Version <<
 514               " SingleInstance:" << Config->SingleInstance <<
 515               " Pipeline:" << Config->Pipeline <<
 516               " SendConfig:" << Config->SendConfig <<
 517               " LocalOnly: " << Config->LocalOnly <<
 518               " NeedsCleanup: " << Config->NeedsCleanup <<
 519               " Removable: " << Config->Removable << endl;
 520    }
 521
 522    return true;
 523 }
 524                                                                         /*}}}*/
 525 // Worker::MediaChange - Request a media change                         /*{{{*/
 526 // ---------------------------------------------------------------------
 527 /* */
 528 bool pkgAcquire::Worker::MediaChange(string Message)
 529 {
 530    int status_fd = _config->FindI("APT::Status-Fd",-1);
 531    if(status_fd > 0)
 532    {
 533       string Media = LookupTag(Message,"Media");
 534       string Drive = LookupTag(Message,"Drive");
 535       ostringstream msg,status;
 536       ioprintf(msg,_("Please insert the disc labeled: "
 537                      "'%s' "
 538                      "in the drive '%s' and press [Enter]."),
 539                Media.c_str(),Drive.c_str());
 540       status << "media-change: "  // message
 541              << Media  << ":"     // media
 542              << Drive  << ":"     // drive
 543              << msg.str()         // l10n message
 544              << endl;
 545
 546       std::string const dlstatus = status.str();
 547       FileFd::Write(status_fd, dlstatus.c_str(), dlstatus.size());
 548    }
 549
 550    if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"),
 551                                     LookupTag(Message,"Drive")) == false)
 552    {
 553       char S[300];
 554       snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
 555       if (Debug == true)
 556          clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
 557       OutQueue += S;
 558       OutReady = true;
 559       return true;
 560    }
 561
 562    char S[300];
 563    snprintf(S,sizeof(S),"603 Media Changed\n\n");
 564    if (Debug == true)
 565       clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
 566    OutQueue += S;
 567    OutReady = true;
 568    return true;
 569 }
 570                                                                         /*}}}*/
 571 // Worker::SendConfiguration - Send the config to the method            /*{{{*/
 572 // ---------------------------------------------------------------------
 573 /* */
 574 bool pkgAcquire::Worker::SendConfiguration()
 575 {
 576    if (Config->SendConfig == false)
 577       return true;
 578
 579    if (OutFd == -1)
 580       return false;
 581
 582    /* Write out all of the configuration directives by walking the
 583       configuration tree */
 584    std::ostringstream Message;
 585    Message << "601 Configuration\n";
 586    _config->Dump(Message, NULL, "Config-Item: %F=%V\n", false);
 587    Message << '\n';
 588
 589    if (Debug == true)
 590       clog << " -> " << Access << ':' << QuoteString(Message.str(),"\n") << endl;
 591    OutQueue += Message.str();
 592    OutReady = true;
 593
 594    return true;
 595 }
 596                                                                         /*}}}*/
 597 // Worker::QueueItem - Add an item to the outbound queue                /*{{{*/
 598 // ---------------------------------------------------------------------
 599 /* Send a URI Acquire message to the method */
 600 bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
 601 {
 602    if (OutFd == -1)
 603       return false;
 604
 605    string Message = "600 URI Acquire\n";
 606    Message.reserve(300);
 607    Message += "URI: " + Item->URI;
 608    Message += "\nFilename: " + Item->Owner->DestFile;
 609
 610    HashStringList const hsl = Item->GetExpectedHashes();
 611    for (HashStringList::const_iterator hs = hsl.begin(); hs != hsl.end(); ++hs)
 612       Message += "\nExpected-" + hs->HashType() + ": " + hs->HashValue();
 613
 614    if (hsl.FileSize() == 0)
 615    {
 616       unsigned long long FileSize = Item->GetMaximumSize();
 617       if(FileSize > 0)
 618       {
 619          string MaximumSize;
 620          strprintf(MaximumSize, "%llu", FileSize);
 621          Message += "\nMaximum-Size: " + MaximumSize;
 622       }
 623    }
 624
 625    Item->SyncDestinationFiles();
 626    Message += Item->Custom600Headers();
 627    Message += "\n\n";
 628
 629    if (RealFileExists(Item->Owner->DestFile))
 630    {
 631       std::string SandboxUser = _config->Find("APT::Sandbox::User");
 632       ChangeOwnerAndPermissionOfFile("Item::QueueURI", Item->Owner->DestFile.c_str(),
 633                                      SandboxUser.c_str(), "root", 0600);
 634    }
 635
 636    if (Debug == true)
 637       clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
 638    OutQueue += Message;
 639    OutReady = true;
 640
 641    return true;
 642 }
 643                                                                         /*}}}*/
 644 // Worker::OutFdRead - Out bound FD is ready                            /*{{{*/
 645 // ---------------------------------------------------------------------
 646 /* */
 647 bool pkgAcquire::Worker::OutFdReady()
 648 {
 649    int Res;
 650    do
 651    {
 652       Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
 653    }
 654    while (Res < 0 && errno == EINTR);
 655
 656    if (Res <= 0)
 657       return MethodFailure();
 658
 659    OutQueue.erase(0,Res);
 660    if (OutQueue.empty() == true)
 661       OutReady = false;
 662
 663    return true;
 664 }
 665                                                                         /*}}}*/
 666 // Worker::InFdRead - In bound FD is ready                              /*{{{*/
 667 // ---------------------------------------------------------------------
 668 /* */
 669 bool pkgAcquire::Worker::InFdReady()
 670 {
 671    if (ReadMessages() == false)
 672       return false;
 673    RunMessages();
 674    return true;
 675 }
 676                                                                         /*}}}*/
 677 // Worker::MethodFailure - Called when the method fails                 /*{{{*/
 678 // ---------------------------------------------------------------------
 679 /* This is called when the method is believed to have failed, probably because
 680    read returned -1. */
 681 bool pkgAcquire::Worker::MethodFailure()
 682 {
 683    _error->Error("Method %s has died unexpectedly!",Access.c_str());
 684
 685    // do not reap the child here to show meaningfull error to the user
 686    ExecWait(Process,Access.c_str(),false);
 687    Process = -1;
 688    close(InFd);
 689    close(OutFd);
 690    InFd = -1;
 691    OutFd = -1;
 692    OutReady = false;
 693    InReady = false;
 694    OutQueue = string();
 695    MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
 696
 697    return false;
 698 }
 699                                                                         /*}}}*/
 700 // Worker::Pulse - Called periodically                                  /*{{{*/
 701 // ---------------------------------------------------------------------
 702 /* */
 703 void pkgAcquire::Worker::Pulse()
 704 {
 705    if (CurrentItem == 0)
 706       return;
 707
 708    struct stat Buf;
 709    if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
 710       return;
 711    CurrentSize = Buf.st_size;
 712 }
 713                                                                         /*}}}*/
 714 // Worker::ItemDone - Called when the current item is finished          /*{{{*/
 715 // ---------------------------------------------------------------------
 716 /* */
 717 void pkgAcquire::Worker::ItemDone()
 718 {
 719    CurrentItem = 0;
 720    CurrentSize = 0;
 721    TotalSize = 0;
 722    Status = string();
 723 }
 724                                                                         /*}}}*/
 725 void pkgAcquire::Worker::PrepareFiles(char const * const caller, pkgAcquire::Queue::QItem const * const Itm)/*{{{*/
 726 {
 727    if (RealFileExists(Itm->Owner->DestFile))
 728    {
 729       ChangeOwnerAndPermissionOfFile(caller, Itm->Owner->DestFile.c_str(), "root", "root", 0644);
 730       std::string const filename = Itm->Owner->DestFile;
 731       for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 732       {
 733          pkgAcquire::Item const * const Owner = *O;
 734          if (Owner->DestFile == filename)
 735             continue;
 736          unlink(Owner->DestFile.c_str());
 737          if (link(filename.c_str(), Owner->DestFile.c_str()) != 0)
 738          {
 739             // different mounts can't happen for us as we download to lists/ by default,
 740             // but if the system is reused by others the locations can potentially be on
 741             // different disks, so use symlink as poor-men replacement.
 742             // FIXME: Real copying as last fallback, but that is costly, so offload to a method preferable
 743             if (symlink(filename.c_str(), Owner->DestFile.c_str()) != 0)
 744                _error->Error("Can't create (sym)link of file %s to %s", filename.c_str(), Owner->DestFile.c_str());
 745          }
 746       }
 747    }
 748    else
 749    {
 750       for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 751          unlink((*O)->DestFile.c_str());
 752    }
 753 }
 754                                                                         /*}}}*/