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