X-Git-Url: https://git.saurik.com/apt.git/blobdiff_plain/93641593cafac296b9072288d8ef9e1a526d745b..2ac6ce927cda2847baf8e71a74e595e6b82c6d98:/apt-pkg/acquire-worker.cc diff --git a/apt-pkg/acquire-worker.cc b/apt-pkg/acquire-worker.cc index 688c5e220..5cd7d6f1e 100644 --- a/apt-pkg/acquire-worker.cc +++ b/apt-pkg/acquire-worker.cc @@ -1,6 +1,6 @@ // -*- mode: cpp; mode: fold -*- // Description /*{{{*/ -// $Id: acquire-worker.cc,v 1.3 1998/10/20 04:33:12 jgg Exp $ +// $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $ /* ###################################################################### Acquire Worker @@ -16,25 +16,40 @@ #pragma implementation "apt-pkg/acquire-worker.h" #endif #include +#include #include #include #include -#include +#include +#include + +#include +#include + +#include #include +#include #include -#include +#include +#include /*}}}*/ +using namespace std; + // Worker::Worker - Constructor for Queue startup /*{{{*/ // --------------------------------------------------------------------- /* */ -pkgAcquire::Worker::Worker(Queue *Q,string Acc) +pkgAcquire::Worker::Worker(Queue *Q,MethodConfig *Cnf, + pkgAcquireStatus *Log) : Log(Log) { OwnerQ = Q; - Config = 0; - Access = Acc; - + Config = Cnf; + Access = Cnf->Access; + CurrentItem = 0; + TotalSize = 0; + CurrentSize = 0; + Construct(); } /*}}}*/ @@ -46,7 +61,10 @@ pkgAcquire::Worker::Worker(MethodConfig *Cnf) OwnerQ = 0; Config = Cnf; Access = Cnf->Access; - + CurrentItem = 0; + TotalSize = 0; + CurrentSize = 0; + Construct(); } /*}}}*/ @@ -55,10 +73,13 @@ pkgAcquire::Worker::Worker(MethodConfig *Cnf) /* */ void pkgAcquire::Worker::Construct() { - Next = 0; + NextQueue = 0; + NextAcquire = 0; Process = -1; InFd = -1; OutFd = -1; + OutReady = false; + InReady = false; Debug = _config->FindB("Debug::pkgAcquire::Worker",false); } /*}}}*/ @@ -71,7 +92,13 @@ pkgAcquire::Worker::~Worker() close(OutFd); if (Process > 0) - kill(Process,SIGINT); + { + /* Closing of stdin is the signal to exit and die when the process + indicates it needs cleanup */ + if (Config->NeedsCleanup == false) + kill(Process,SIGINT); + ExecWait(Process,Access.c_str(),true); + } } /*}}}*/ // Worker::Start - Start the worker process /*{{{*/ @@ -82,7 +109,7 @@ bool pkgAcquire::Worker::Start() // Get the method path string Method = _config->FindDir("Dir::Bin::Methods") + Access; if (FileExists(Method) == false) - return _error->Error("The method driver %s could not be found.",Method.c_str()); + return _error->Error(_("The method driver %s could not be found."),Method.c_str()); if (Debug == true) clog << "Starting method '" << Method << '\'' << endl; @@ -96,24 +123,16 @@ bool pkgAcquire::Worker::Start() close(Pipes[I]); return false; } - + for (int I = 0; I != 4; I++) + SetCloseExec(Pipes[I],true); + // Fork off the process - Process = fork(); - if (Process < 0) - { - cerr << "FATAL -> Failed to fork." << endl; - exit(100); - } - - // Spawn the subprocess + Process = ExecFork(); if (Process == 0) { // Setup the FDs dup2(Pipes[1],STDOUT_FILENO); dup2(Pipes[2],STDIN_FILENO); - dup2(((filebuf *)clog.rdbuf())->fd(),STDERR_FILENO); - for (int I = 0; I != 4; I++) - close(Pipes[I]); SetCloseExec(STDOUT_FILENO,false); SetCloseExec(STDIN_FILENO,false); SetCloseExec(STDERR_FILENO,false); @@ -123,7 +142,7 @@ bool pkgAcquire::Worker::Start() Args[1] = 0; execv(Args[0],(char **)Args); cerr << "Failed to exec method " << Args[0] << endl; - exit(100); + _exit(100); } // Fix up our FDs @@ -133,81 +152,31 @@ bool pkgAcquire::Worker::Start() SetNonBlock(Pipes[3],true); close(Pipes[1]); close(Pipes[2]); + OutReady = false; + InReady = true; // Read the configuration data if (WaitFd(InFd) == false || ReadMessages() == false) - return _error->Error("Method %s did not start correctly",Method.c_str()); + return _error->Error(_("Method %s did not start correctly"),Method.c_str()); RunMessages(); + if (OwnerQ != 0) + SendConfiguration(); return true; } /*}}}*/ // Worker::ReadMessages - Read all pending messages into the list /*{{{*/ // --------------------------------------------------------------------- -/* This pulls full messages from the input FD into the message buffer. - It assumes that messages will not pause during transit so no - fancy buffering is used. */ +/* */ bool pkgAcquire::Worker::ReadMessages() { - char Buffer[4000]; - char *End = Buffer; - - while (1) - { - int Res = read(InFd,End,sizeof(Buffer) - (End-Buffer)); - - // Process is dead, this is kind of bad.. - if (Res == 0) - { - if (waitpid(Process,0,0) != Process) - _error->Warning("I waited but nothing was there!"); - Process = -1; - close(InFd); - close(OutFd); - InFd = -1; - OutFd = -1; - return false; - } - - // No data - if (Res == -1) - return true; - - End += Res; - - // Look for the end of the message - for (char *I = Buffer; I < End; I++) - { - if (I[0] != '\n' || I[1] != '\n') - continue; - - // Pull the message out - string Message(Buffer,0,I-Buffer); - - // Fix up the buffer - for (; I < End && *I == '\n'; I++); - End -= I-Buffer; - memmove(Buffer,I,End-Buffer); - I = Buffer; - - if (Debug == true) - clog << "Message " << Access << ':' << QuoteString(Message,"\n") << endl; - - MessageQueue.push_back(Message); - } - if (End == Buffer) - return true; - - if (WaitFd(InFd) == false) - return false; - } - + if (::ReadMessages(InFd,MessageQueue) == false) + return MethodFailure(); return true; } /*}}}*/ - // Worker::RunMessage - Empty the message queue /*{{{*/ // --------------------------------------------------------------------- /* This takes the messages from the message queue and runs them through @@ -218,6 +187,9 @@ bool pkgAcquire::Worker::RunMessages() { string Message = MessageQueue.front(); MessageQueue.erase(MessageQueue.begin()); + + if (Debug == true) + clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl; // Fetch the message number char *End; @@ -225,13 +197,133 @@ bool pkgAcquire::Worker::RunMessages() if (End == Message.c_str()) return _error->Error("Invalid message from method %s: %s",Access.c_str(),Message.c_str()); + string URI = LookupTag(Message,"URI"); + pkgAcquire::Queue::QItem *Itm = 0; + if (URI.empty() == false) + Itm = OwnerQ->FindItem(URI,this); + // Determine the message number and dispatch switch (Number) { + // 100 Capabilities case 100: if (Capabilities(Message) == false) return _error->Error("Unable to process Capabilities message from %s",Access.c_str()); break; + + // 101 Log + case 101: + if (Debug == true) + clog << " <- (log) " << LookupTag(Message,"Message") << endl; + break; + + // 102 Status + case 102: + Status = LookupTag(Message,"Message"); + break; + + // 200 URI Start + case 200: + { + if (Itm == 0) + { + _error->Error("Method gave invalid 200 URI Start message"); + break; + } + + CurrentItem = Itm; + CurrentSize = 0; + TotalSize = atoi(LookupTag(Message,"Size","0").c_str()); + ResumePoint = atoi(LookupTag(Message,"Resume-Point","0").c_str()); + Itm->Owner->Start(Message,atoi(LookupTag(Message,"Size","0").c_str())); + + // Display update before completion + if (Log != 0 && Log->MorePulses == true) + Log->Pulse(Itm->Owner->GetOwner()); + + if (Log != 0) + Log->Fetch(*Itm); + + break; + } + + // 201 URI Done + case 201: + { + if (Itm == 0) + { + _error->Error("Method gave invalid 201 URI Done message"); + break; + } + + pkgAcquire::Item *Owner = Itm->Owner; + pkgAcquire::ItemDesc Desc = *Itm; + + // Display update before completion + if (Log != 0 && Log->MorePulses == true) + Log->Pulse(Owner->GetOwner()); + + OwnerQ->ItemDone(Itm); + if (TotalSize != 0 && + (unsigned)atoi(LookupTag(Message,"Size","0").c_str()) != TotalSize) + _error->Warning("Bizarre Error - File size is not what the server reported %s %lu", + LookupTag(Message,"Size","0").c_str(),TotalSize); + + Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()), + LookupTag(Message,"MD5-Hash"),Config); + ItemDone(); + + // Log that we are done + if (Log != 0) + { + if (StringToBool(LookupTag(Message,"IMS-Hit"),false) == true || + StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false) == true) + { + /* Hide 'hits' for local only sources - we also manage to + hide gets */ + if (Config->LocalOnly == false) + Log->IMSHit(Desc); + } + else + Log->Done(Desc); + } + break; + } + + // 400 URI Failure + case 400: + { + if (Itm == 0) + { + _error->Error("Method gave invalid 400 URI Failure message"); + break; + } + + // Display update before completion + if (Log != 0 && Log->MorePulses == true) + Log->Pulse(Itm->Owner->GetOwner()); + + pkgAcquire::Item *Owner = Itm->Owner; + pkgAcquire::ItemDesc Desc = *Itm; + OwnerQ->ItemDone(Itm); + Owner->Failed(Message,Config); + ItemDone(); + + if (Log != 0) + Log->Fail(Desc); + + break; + } + + // 401 General Failure + case 401: + _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str()); + break; + + // 403 Media Change + case 403: + MediaChange(Message); + break; } } return true; @@ -248,16 +340,209 @@ bool pkgAcquire::Worker::Capabilities(string Message) Config->Version = LookupTag(Message,"Version"); Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false); - Config->PreScan = StringToBool(LookupTag(Message,"Pre-Scan"),false); + Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false); + Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false); + Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false); + Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false); + Config->Removable = StringToBool(LookupTag(Message,"Removable"),false); // Some debug text if (Debug == true) { clog << "Configured access method " << Config->Access << endl; - clog << "Version: " << Config->Version << " SingleInstance: " << - Config->SingleInstance << " PreScan: " << Config->PreScan << endl; + clog << "Version:" << Config->Version << + " SingleInstance:" << Config->SingleInstance << + " Pipeline:" << Config->Pipeline << + " SendConfig:" << Config->SendConfig << + " LocalOnly: " << Config->LocalOnly << + " NeedsCleanup: " << Config->NeedsCleanup << + " Removable: " << Config->Removable << endl; + } + + return true; +} + /*}}}*/ +// Worker::MediaChange - Request a media change /*{{{*/ +// --------------------------------------------------------------------- +/* */ +bool pkgAcquire::Worker::MediaChange(string Message) +{ + if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"), + LookupTag(Message,"Drive")) == false) + { + char S[300]; + snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n"); + if (Debug == true) + clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl; + OutQueue += S; + OutReady = true; + return true; + } + + char S[300]; + snprintf(S,sizeof(S),"603 Media Changed\n\n"); + if (Debug == true) + clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl; + OutQueue += S; + OutReady = true; + return true; +} + /*}}}*/ +// Worker::SendConfiguration - Send the config to the method /*{{{*/ +// --------------------------------------------------------------------- +/* */ +bool pkgAcquire::Worker::SendConfiguration() +{ + if (Config->SendConfig == false) + return true; + + if (OutFd == -1) + return false; + + string Message = "601 Configuration\n"; + Message.reserve(2000); + + /* Write out all of the configuration directives by walking the + configuration tree */ + const Configuration::Item *Top = _config->Tree(0); + for (; Top != 0;) + { + if (Top->Value.empty() == false) + { + string Line = "Config-Item: " + QuoteString(Top->FullTag(),"=\"\n") + "="; + Line += QuoteString(Top->Value,"\n") + '\n'; + Message += Line; + } + + if (Top->Child != 0) + { + Top = Top->Child; + continue; + } + + while (Top != 0 && Top->Next == 0) + Top = Top->Parent; + if (Top != 0) + Top = Top->Next; + } + Message += '\n'; + + if (Debug == true) + clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl; + OutQueue += Message; + OutReady = true; + + return true; +} + /*}}}*/ +// Worker::QueueItem - Add an item to the outbound queue /*{{{*/ +// --------------------------------------------------------------------- +/* Send a URI Acquire message to the method */ +bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item) +{ + if (OutFd == -1) + return false; + + string Message = "600 URI Acquire\n"; + Message.reserve(300); + Message += "URI: " + Item->URI; + Message += "\nFilename: " + Item->Owner->DestFile; + Message += Item->Owner->Custom600Headers(); + Message += "\n\n"; + + if (Debug == true) + clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl; + OutQueue += Message; + OutReady = true; + + return true; +} + /*}}}*/ +// Worker::OutFdRead - Out bound FD is ready /*{{{*/ +// --------------------------------------------------------------------- +/* */ +bool pkgAcquire::Worker::OutFdReady() +{ + int Res; + do + { + Res = write(OutFd,OutQueue.c_str(),OutQueue.length()); } + while (Res < 0 && errno == EINTR); + + if (Res <= 0) + return MethodFailure(); + + // Hmm.. this should never happen. + if (Res < 0) + return true; + + OutQueue.erase(0,Res); + if (OutQueue.empty() == true) + OutReady = false; return true; } /*}}}*/ +// Worker::InFdRead - In bound FD is ready /*{{{*/ +// --------------------------------------------------------------------- +/* */ +bool pkgAcquire::Worker::InFdReady() +{ + if (ReadMessages() == false) + return false; + RunMessages(); + return true; +} + /*}}}*/ +// Worker::MethodFailure - Called when the method fails /*{{{*/ +// --------------------------------------------------------------------- +/* This is called when the method is belived to have failed, probably because + read returned -1. */ +bool pkgAcquire::Worker::MethodFailure() +{ + _error->Error("Method %s has died unexpectedly!",Access.c_str()); + + ExecWait(Process,Access.c_str(),true); + Process = -1; + close(InFd); + close(OutFd); + InFd = -1; + OutFd = -1; + OutReady = false; + InReady = false; + OutQueue = string(); + MessageQueue.erase(MessageQueue.begin(),MessageQueue.end()); + + return false; +} + /*}}}*/ +// Worker::Pulse - Called periodically /*{{{*/ +// --------------------------------------------------------------------- +/* */ +void pkgAcquire::Worker::Pulse() +{ + if (CurrentItem == 0) + return; + + struct stat Buf; + if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0) + return; + CurrentSize = Buf.st_size; + + // Hmm? Should not happen... + if (CurrentSize > TotalSize && TotalSize != 0) + TotalSize = CurrentSize; +} + /*}}}*/ +// Worker::ItemDone - Called when the current item is finished /*{{{*/ +// --------------------------------------------------------------------- +/* */ +void pkgAcquire::Worker::ItemDone() +{ + CurrentItem = 0; + CurrentSize = 0; + TotalSize = 0; + Status = string(); +} + /*}}}*/