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             HashStringList ReceivedHashes;
 339             {
 340                std::string const givenfilename = LookupTag(Message, "Filename");
 341                std::string const filename = givenfilename.empty() ? Itm->Owner->DestFile : givenfilename;
 342                // see if we got hashes to verify
 343                for (char const * const * type = HashString::SupportedHashes(); *type != NULL; ++type)
 344                {
 345                   std::string const tagname = std::string(*type) + "-Hash";
 346                   std::string const hashsum = LookupTag(Message, tagname.c_str());
 347                   if (hashsum.empty() == false)
 348                      ReceivedHashes.push_back(HashString(*type, hashsum));
 349                }
 350                // not all methods always sent Hashes our way
 351                if (ReceivedHashes.usable() == false)
 352                {
 353                   HashStringList const ExpectedHashes = Itm->GetExpectedHashes();
 354                   if (ExpectedHashes.usable() == true && RealFileExists(filename))
 355                   {
 356                      Hashes calc(ExpectedHashes);
 357                      FileFd file(filename, FileFd::ReadOnly, FileFd::None);
 358                      calc.AddFD(file);
 359                      ReceivedHashes = calc.GetHashStringList();
 360                   }
 361                }
 362
 363                // only local files can refer other filenames and counting them as fetched would be unfair
 364                if (Log != NULL && Itm->Owner->Complete == false && Itm->Owner->Local == false && givenfilename == filename)
 365                   Log->Fetched(ReceivedHashes.FileSize(),atoi(LookupTag(Message,"Resume-Point","0").c_str()));
 366             }
 367
 368             std::vector<Item*> const ItmOwners = Itm->Owners;
 369             OwnerQ->ItemDone(Itm);
 370             Itm = NULL;
 371
 372             bool const isIMSHit = StringToBool(LookupTag(Message,"IMS-Hit"),false) ||
 373                StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false);
 374             for (pkgAcquire::Queue::QItem::owner_iterator O = ItmOwners.begin(); O != ItmOwners.end(); ++O)
 375             {
 376                pkgAcquire::Item * const Owner = *O;
 377                HashStringList const ExpectedHashes = Owner->GetExpectedHashes();
 378                if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
 379                {
 380                   std::clog << "201 URI Done: " << Owner->DescURI() << endl
 381                      << "ReceivedHash:" << endl;
 382                   for (HashStringList::const_iterator hs = ReceivedHashes.begin(); hs != ReceivedHashes.end(); ++hs)
 383                      std::clog <<  "\t- " << hs->toStr() << std::endl;
 384                   std::clog << "ExpectedHash:" << endl;
 385                   for (HashStringList::const_iterator hs = ExpectedHashes.begin(); hs != ExpectedHashes.end(); ++hs)
 386                      std::clog <<  "\t- " << hs->toStr() << std::endl;
 387                   std::clog << endl;
 388                }
 389
 390                // decide if what we got is what we expected
 391                bool consideredOkay = false;
 392                if (ExpectedHashes.usable())
 393                {
 394                   if (ReceivedHashes.usable() == false)
 395                   {
 396                      /* IMS-Hits can't be checked here as we will have uncompressed file,
 397                         but the hashes for the compressed file. What we have was good through
 398                         so all we have to ensure later is that we are not stalled. */
 399                      consideredOkay = isIMSHit;
 400                   }
 401                   else if (ReceivedHashes == ExpectedHashes)
 402                      consideredOkay = true;
 403                   else
 404                      consideredOkay = false;
 405
 406                }
 407                else if (Owner->HashesRequired() == true)
 408                   consideredOkay = false;
 409                else
 410                   consideredOkay = true;
 411
 412                if (consideredOkay == true)
 413                   consideredOkay = Owner->VerifyDone(Message, Config);
 414                else // hashsum mismatch
 415                   Owner->Status = pkgAcquire::Item::StatAuthError;
 416
 417                if (consideredOkay == true)
 418                {
 419                   Owner->Done(Message, ReceivedHashes, Config);
 420                   if (Log != 0)
 421                   {
 422                      if (isIMSHit)
 423                         Log->IMSHit(Owner->GetItemDesc());
 424                      else
 425                         Log->Done(Owner->GetItemDesc());
 426                   }
 427                }
 428                else
 429                {
 430                   Owner->Failed(Message,Config);
 431                   if (Log != 0)
 432                      Log->Fail(Owner->GetItemDesc());
 433                }
 434             }
 435             ItemDone();
 436             break;
 437          }
 438
 439          // 400 URI Failure
 440          case 400:
 441          {
 442             if (Itm == 0)
 443             {
 444                std::string const msg = LookupTag(Message,"Message");
 445                _error->Error("Method gave invalid 400 URI Failure message: %s", msg.c_str());
 446                break;
 447             }
 448
 449             PrepareFiles("400::URIFailure", Itm);
 450
 451             // Display update before completion
 452             if (Log != 0 && Log->MorePulses == true)
 453                for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 454                   Log->Pulse((*O)->GetOwner());
 455
 456             std::vector<Item*> const ItmOwners = Itm->Owners;
 457             OwnerQ->ItemDone(Itm);
 458             Itm = NULL;
 459
 460             for (pkgAcquire::Queue::QItem::owner_iterator O = ItmOwners.begin(); O != ItmOwners.end(); ++O)
 461             {
 462                // set some status
 463                if(LookupTag(Message,"FailReason") == "Timeout" ||
 464                      LookupTag(Message,"FailReason") == "TmpResolveFailure" ||
 465                      LookupTag(Message,"FailReason") == "ResolveFailure" ||
 466                      LookupTag(Message,"FailReason") == "ConnectionRefused")
 467                   (*O)->Status = pkgAcquire::Item::StatTransientNetworkError;
 468
 469                (*O)->Failed(Message,Config);
 470
 471                if (Log != 0)
 472                   Log->Fail((*O)->GetItemDesc());
 473             }
 474             ItemDone();
 475
 476             break;
 477          }
 478
 479          // 401 General Failure
 480          case 401:
 481          _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
 482          break;
 483
 484          // 403 Media Change
 485          case 403:
 486          MediaChange(Message);
 487          break;
 488       }
 489    }
 490    return true;
 491 }
 492                                                                         /*}}}*/
 493 // Worker::Capabilities - 100 Capabilities handler                      /*{{{*/
 494 // ---------------------------------------------------------------------
 495 /* This parses the capabilities message and dumps it into the configuration
 496    structure. */
 497 bool pkgAcquire::Worker::Capabilities(string Message)
 498 {
 499    if (Config == 0)
 500       return true;
 501
 502    Config->Version = LookupTag(Message,"Version");
 503    Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
 504    Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
 505    Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
 506    Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
 507    Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
 508    Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);
 509
 510    // Some debug text
 511    if (Debug == true)
 512    {
 513       clog << "Configured access method " << Config->Access << endl;
 514       clog << "Version:" << Config->Version <<
 515               " SingleInstance:" << Config->SingleInstance <<
 516               " Pipeline:" << Config->Pipeline <<
 517               " SendConfig:" << Config->SendConfig <<
 518               " LocalOnly: " << Config->LocalOnly <<
 519               " NeedsCleanup: " << Config->NeedsCleanup <<
 520               " Removable: " << Config->Removable << endl;
 521    }
 522
 523    return true;
 524 }
 525                                                                         /*}}}*/
 526 // Worker::MediaChange - Request a media change                         /*{{{*/
 527 // ---------------------------------------------------------------------
 528 /* */
 529 bool pkgAcquire::Worker::MediaChange(string Message)
 530 {
 531    int status_fd = _config->FindI("APT::Status-Fd",-1);
 532    if(status_fd > 0)
 533    {
 534       string Media = LookupTag(Message,"Media");
 535       string Drive = LookupTag(Message,"Drive");
 536       ostringstream msg,status;
 537       ioprintf(msg,_("Please insert the disc labeled: "
 538                      "'%s' "
 539                      "in the drive '%s' and press [Enter]."),
 540                Media.c_str(),Drive.c_str());
 541       status << "media-change: "  // message
 542              << Media  << ":"     // media
 543              << Drive  << ":"     // drive
 544              << msg.str()         // l10n message
 545              << endl;
 546
 547       std::string const dlstatus = status.str();
 548       FileFd::Write(status_fd, dlstatus.c_str(), dlstatus.size());
 549    }
 550
 551    if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"),
 552                                     LookupTag(Message,"Drive")) == false)
 553    {
 554       char S[300];
 555       snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
 556       if (Debug == true)
 557          clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
 558       OutQueue += S;
 559       OutReady = true;
 560       return true;
 561    }
 562
 563    char S[300];
 564    snprintf(S,sizeof(S),"603 Media Changed\n\n");
 565    if (Debug == true)
 566       clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
 567    OutQueue += S;
 568    OutReady = true;
 569    return true;
 570 }
 571                                                                         /*}}}*/
 572 // Worker::SendConfiguration - Send the config to the method            /*{{{*/
 573 // ---------------------------------------------------------------------
 574 /* */
 575 bool pkgAcquire::Worker::SendConfiguration()
 576 {
 577    if (Config->SendConfig == false)
 578       return true;
 579
 580    if (OutFd == -1)
 581       return false;
 582
 583    /* Write out all of the configuration directives by walking the
 584       configuration tree */
 585    std::ostringstream Message;
 586    Message << "601 Configuration\n";
 587    _config->Dump(Message, NULL, "Config-Item: %F=%V\n", false);
 588    Message << '\n';
 589
 590    if (Debug == true)
 591       clog << " -> " << Access << ':' << QuoteString(Message.str(),"\n") << endl;
 592    OutQueue += Message.str();
 593    OutReady = true;
 594
 595    return true;
 596 }
 597                                                                         /*}}}*/
 598 // Worker::QueueItem - Add an item to the outbound queue                /*{{{*/
 599 // ---------------------------------------------------------------------
 600 /* Send a URI Acquire message to the method */
 601 bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
 602 {
 603    if (OutFd == -1)
 604       return false;
 605
 606    string Message = "600 URI Acquire\n";
 607    Message.reserve(300);
 608    Message += "URI: " + Item->URI;
 609    Message += "\nFilename: " + Item->Owner->DestFile;
 610
 611    HashStringList const hsl = Item->GetExpectedHashes();
 612    for (HashStringList::const_iterator hs = hsl.begin(); hs != hsl.end(); ++hs)
 613       Message += "\nExpected-" + hs->HashType() + ": " + hs->HashValue();
 614
 615    if (hsl.FileSize() == 0)
 616    {
 617       unsigned long long FileSize = Item->GetMaximumSize();
 618       if(FileSize > 0)
 619       {
 620          string MaximumSize;
 621          strprintf(MaximumSize, "%llu", FileSize);
 622          Message += "\nMaximum-Size: " + MaximumSize;
 623       }
 624    }
 625
 626    Item->SyncDestinationFiles();
 627    Message += Item->Custom600Headers();
 628    Message += "\n\n";
 629
 630    if (RealFileExists(Item->Owner->DestFile))
 631    {
 632       std::string SandboxUser = _config->Find("APT::Sandbox::User");
 633       ChangeOwnerAndPermissionOfFile("Item::QueueURI", Item->Owner->DestFile.c_str(),
 634                                      SandboxUser.c_str(), "root", 0600);
 635    }
 636
 637    if (Debug == true)
 638       clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
 639    OutQueue += Message;
 640    OutReady = true;
 641
 642    return true;
 643 }
 644                                                                         /*}}}*/
 645 // Worker::OutFdRead - Out bound FD is ready                            /*{{{*/
 646 // ---------------------------------------------------------------------
 647 /* */
 648 bool pkgAcquire::Worker::OutFdReady()
 649 {
 650    int Res;
 651    do
 652    {
 653       Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
 654    }
 655    while (Res < 0 && errno == EINTR);
 656
 657    if (Res <= 0)
 658       return MethodFailure();
 659
 660    OutQueue.erase(0,Res);
 661    if (OutQueue.empty() == true)
 662       OutReady = false;
 663
 664    return true;
 665 }
 666                                                                         /*}}}*/
 667 // Worker::InFdRead - In bound FD is ready                              /*{{{*/
 668 // ---------------------------------------------------------------------
 669 /* */
 670 bool pkgAcquire::Worker::InFdReady()
 671 {
 672    if (ReadMessages() == false)
 673       return false;
 674    RunMessages();
 675    return true;
 676 }
 677                                                                         /*}}}*/
 678 // Worker::MethodFailure - Called when the method fails                 /*{{{*/
 679 // ---------------------------------------------------------------------
 680 /* This is called when the method is believed to have failed, probably because
 681    read returned -1. */
 682 bool pkgAcquire::Worker::MethodFailure()
 683 {
 684    _error->Error("Method %s has died unexpectedly!",Access.c_str());
 685
 686    // do not reap the child here to show meaningfull error to the user
 687    ExecWait(Process,Access.c_str(),false);
 688    Process = -1;
 689    close(InFd);
 690    close(OutFd);
 691    InFd = -1;
 692    OutFd = -1;
 693    OutReady = false;
 694    InReady = false;
 695    OutQueue = string();
 696    MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
 697
 698    return false;
 699 }
 700                                                                         /*}}}*/
 701 // Worker::Pulse - Called periodically                                  /*{{{*/
 702 // ---------------------------------------------------------------------
 703 /* */
 704 void pkgAcquire::Worker::Pulse()
 705 {
 706    if (CurrentItem == 0)
 707       return;
 708
 709    struct stat Buf;
 710    if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
 711       return;
 712    CurrentSize = Buf.st_size;
 713 }
 714                                                                         /*}}}*/
 715 // Worker::ItemDone - Called when the current item is finished          /*{{{*/
 716 // ---------------------------------------------------------------------
 717 /* */
 718 void pkgAcquire::Worker::ItemDone()
 719 {
 720    CurrentItem = 0;
 721    CurrentSize = 0;
 722    TotalSize = 0;
 723    Status = string();
 724 }
 725                                                                         /*}}}*/
 726 void pkgAcquire::Worker::PrepareFiles(char const * const caller, pkgAcquire::Queue::QItem const * const Itm)/*{{{*/
 727 {
 728    if (RealFileExists(Itm->Owner->DestFile))
 729    {
 730       ChangeOwnerAndPermissionOfFile(caller, Itm->Owner->DestFile.c_str(), "root", "root", 0644);
 731       std::string const filename = Itm->Owner->DestFile;
 732       for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 733       {
 734          pkgAcquire::Item const * const Owner = *O;
 735          if (Owner->DestFile == filename)
 736             continue;
 737          unlink(Owner->DestFile.c_str());
 738          if (link(filename.c_str(), Owner->DestFile.c_str()) != 0)
 739          {
 740             // different mounts can't happen for us as we download to lists/ by default,
 741             // but if the system is reused by others the locations can potentially be on
 742             // different disks, so use symlink as poor-men replacement.
 743             // FIXME: Real copying as last fallback, but that is costly, so offload to a method preferable
 744             if (symlink(filename.c_str(), Owner->DestFile.c_str()) != 0)
 745                _error->Error("Can't create (sym)link of file %s to %s", filename.c_str(), Owner->DestFile.c_str());
 746          }
 747       }
 748    }
 749    else
 750    {
 751       for (pkgAcquire::Queue::QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 752          unlink((*O)->DestFile.c_str());
 753    }
 754 }
 755                                                                         /*}}}*/