Merge branch 'pu/fetch-at' into 'main'

Main-process-side implementation of retry back-off

See merge request apt-team/apt!181

5 files changed, 104 insertions, 14 deletions

diff --git a/apt-pkg/acquire-item.cc b/apt-pkg/acquire-item.cc
index 37a715390..eb8053feb 100644
--- a/apt-pkg/acquire-item.cc
+++ b/apt-pkg/acquire-item.cc
@@ -290,6 +290,7 @@ public:
    std::vector<std::string> BadAlternativeSites;
    std::vector<std::string> PastRedirections;
    std::unordered_map<std::string, std::string> CustomFields;
+   time_point FetchAfter = {};
 
    Private()
    {
@@ -1120,6 +1121,15 @@ std::string pkgAcquire::Item::HashSum() const				/*{{{*/
    return hs != NULL ? hs->toStr() : "";
 }
 									/*}}}*/
+void pkgAcquire::Item::FetchAfter(time_point FetchAfter) /*{{{*/
+{
+   d->FetchAfter = FetchAfter;
+}
+pkgAcquire::time_point pkgAcquire::Item::FetchAfter()
+{
+   return d->FetchAfter;
+}
+									/*}}}*/
 bool pkgAcquire::Item::IsRedirectionLoop(std::string const &NewURI)	/*{{{*/
 {
    // store can fail due to permission errors and the item will "loop" then
@@ -3205,8 +3215,8 @@ bool pkgAcqIndex::CommonFailed(std::string const &TargetURI,
    {
       if (CompressionExtensions.empty() == false)
       {
-	 Init(TargetURI, Desc.Description, Desc.ShortDesc);
 	 Status = StatIdle;
+	 Init(TargetURI, Desc.Description, Desc.ShortDesc);
 	 return true;
       }
    }
diff --git a/apt-pkg/acquire-item.h b/apt-pkg/acquire-item.h
index 3be8a9c62..22da9815d 100644
--- a/apt-pkg/acquire-item.h
+++ b/apt-pkg/acquire-item.h
@@ -301,6 +301,10 @@ class APT_PUBLIC pkgAcquire::Item : public WeakPointable				/*{{{*/
    /** \brief The priority of the item, used for queuing */
    int APT_HIDDEN Priority();
 
+   /** \brief internal clock definitions to avoid typing all that all over the place */
+   void APT_HIDDEN FetchAfter(time_point FetchAfter);
+   time_point APT_HIDDEN FetchAfter();
+
    protected:
    /** \brief The acquire object with which this item is associated. */
    pkgAcquire * const Owner;
diff --git a/apt-pkg/acquire-worker.cc b/apt-pkg/acquire-worker.cc
index 3fbc5c09f..1bf07e82a 100644
--- a/apt-pkg/acquire-worker.cc
+++ b/apt-pkg/acquire-worker.cc
@@ -632,6 +632,7 @@ void pkgAcquire::Worker::HandleFailure(std::vector<pkgAcquire::Item *> const &It
 				       pkgAcquire::MethodConfig *const Config, pkgAcquireStatus *const Log,
 				       std::string const &Message, bool const errTransient, bool const errAuthErr)
 {
+   auto currentTime = clock::now();
    for (auto const Owner : ItmOwners)
    {
       std::string NewURI;
@@ -640,6 +641,20 @@ void pkgAcquire::Worker::HandleFailure(std::vector<pkgAcquire::Item *> const &It
 	 --Owner->Retries;
 	 Owner->FailMessage(Message);
 	 auto SavedDesc = Owner->GetItemDesc();
+	 if (_config->FindB("Acquire::Retries::Delay", true))
+	 {
+	    auto Iter = _config->FindI("Acquire::Retries", 3) - Owner->Retries - 1;
+	    auto const MaxDur = _config->FindI("Acquire::Retries::Delay::Maximum", 30);
+	    auto Dur =  std::chrono::seconds(std::min(1 << Iter, MaxDur));
+	    if (_config->FindB("Debug::Acquire::Retries"))
+	       std::clog << "Delaying " << SavedDesc.Description << " by " << Dur.count() << " seconds" << std::endl;
+	    Owner->FetchAfter(currentTime + Dur);
+	 }
+	 else
+	 {
+	    Owner->FetchAfter(currentTime);
+	 }
+
 	 if (Log != nullptr)
 	    Log->Fail(SavedDesc);
 	 if (isDoomedItem(Owner) == false)
diff --git a/apt-pkg/acquire.cc b/apt-pkg/acquire.cc
index 8693de930..c3682504f 100644
--- a/apt-pkg/acquire.cc
+++ b/apt-pkg/acquire.cc
@@ -24,15 +24,17 @@
 
 #include <algorithm>
 #include <chrono>
+#include <cmath>
 #include <iomanip>
 #include <iostream>
 #include <memory>
 #include <numeric>
 #include <sstream>
 #include <string>
+#include <tuple>
 #include <vector>
-#include <cmath>
 
+#include <assert.h>
 #include <dirent.h>
 #include <errno.h>
 #include <fcntl.h>
@@ -51,6 +53,14 @@
 
 using namespace std;
 
+// helper to convert time_point to a timeval
+static struct timeval SteadyDurationToTimeVal(std::chrono::steady_clock::duration Time)
+{
+   auto const Time_sec = std::chrono::duration_cast<std::chrono::seconds>(Time);
+   auto const Time_usec = std::chrono::duration_cast<std::chrono::microseconds>(Time - Time_sec);
+   return {Time_sec.count(), Time_usec.count()};
+}
+
 std::string pkgAcquire::URIEncode(std::string const &part)		/*{{{*/
 {
    // The "+" is encoded as a workaround for an S3 bug (LP#1003633 and LP#1086997)
@@ -686,7 +696,7 @@ static void CheckDropPrivsMustBeDisabled(pkgAcquire const &Fetcher)
    if (setgroups(old_gidlist_nr, old_gidlist.get()))
       _error->FatalE("setgroups", "setgroups %u failed", 0);
 }
-pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
+pkgAcquire::RunResult pkgAcquire::Run(int PulseInterval)
 {
    _error->PushToStack();
    CheckDropPrivsMustBeDisabled(*this);
@@ -702,9 +712,7 @@ pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
    bool WasCancelled = false;
 
    // Run till all things have been acquired
-   struct timeval tv;
-   tv.tv_sec = 0;
-   tv.tv_usec = PulseIntervall; 
+   struct timeval tv = SteadyDurationToTimeVal(std::chrono::microseconds(PulseInterval));
    while (ToFetch > 0)
    {
       fd_set RFds;
@@ -713,7 +721,37 @@ pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
       FD_ZERO(&RFds);
       FD_ZERO(&WFds);
       SetFds(Highest,&RFds,&WFds);
-      
+
+      // Shorten the select() cycle in case we have items about to become ready
+      auto now = clock::now();
+      auto fetchAfter = time_point{};
+      for (Queue *I = Queues; I != nullptr; I = I->Next)
+      {
+	 if (I->Items == nullptr)
+	    continue;
+
+	 auto f = I->Items->GetFetchAfter();
+
+	 if (f == time_point() || I->Items->Owner->Status != pkgAcquire::Item::StatIdle)
+	    continue;
+
+	 if (f <= now)
+	 {
+	    I->Cycle();	      // Queue got stuck, unstuck it.
+	    fetchAfter = now; // need to time out in select() below
+	    assert(I->Items->Owner->Status != pkgAcquire::Item::StatIdle);
+	 }
+	 else if (f < fetchAfter || fetchAfter == time_point{})
+	 {
+	    fetchAfter = f;
+	 }
+      }
+
+      if (fetchAfter != time_point{} && (fetchAfter - now) < std::chrono::seconds(tv.tv_sec) + std::chrono::microseconds(tv.tv_usec))
+      {
+	 tv = SteadyDurationToTimeVal(fetchAfter - now);
+      }
+
       int Res;
       do
       {
@@ -733,7 +771,8 @@ pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
       // Timeout, notify the log class
       if (Res == 0 || (Log != 0 && Log->Update == true))
       {
-	 tv.tv_usec = PulseIntervall;
+	 tv = SteadyDurationToTimeVal(std::chrono::microseconds(PulseInterval));
+
 	 for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 	    I->Pulse();
 	 if (Log != 0 && Log->Pulse(this) == false)
@@ -962,6 +1001,7 @@ bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
    };
    QItem **OptimalI = &Items;
    QItem **I = &Items;
+   auto insertLocation = std::make_tuple(Item.Owner->FetchAfter(), -Item.Owner->Priority());
    // move to the end of the queue and check for duplicates here
    for (; *I != 0; ) {
       if (Item.URI == (*I)->URI && MetaKeysMatch(Item, *I))
@@ -974,10 +1014,12 @@ bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
       }
       // Determine the optimal position to insert: before anything with a
       // higher priority.
-      int priority = (*I)->GetPriority();
+      auto queueLocation = std::make_tuple((*I)->GetFetchAfter(),
+					   -(*I)->GetPriority());
 
       I = &(*I)->Next;
-      if (priority >= Item.Owner->Priority()) {
+      if (queueLocation <= insertLocation)
+      {
 	 OptimalI = I;
       }
    }
@@ -1153,6 +1195,7 @@ bool pkgAcquire::Queue::Cycle()
    // Look for a queable item
    QItem *I = Items;
    int ActivePriority = 0;
+   auto currentTime = clock::now();
    while (PipeDepth < static_cast<decltype(PipeDepth)>(MaxPipeDepth))
    {
       for (; I != 0; I = I->Next) {
@@ -1170,6 +1213,11 @@ bool pkgAcquire::Queue::Cycle()
       // the queue is idle
       if (I->GetPriority() < ActivePriority)
 	 return true;
+
+      // Item is not ready yet, delay
+      if (I->GetFetchAfter() > currentTime)
+	 return true;
+
       I->Worker = Workers;
       for (auto const &O: I->Owners)
 	 O->Status = pkgAcquire::Item::StatFetching;
@@ -1237,6 +1285,15 @@ APT_PURE int pkgAcquire::Queue::QItem::GetPriority() const		/*{{{*/
    return Priority;
 }
 									/*}}}*/
+APT_PURE pkgAcquire::time_point pkgAcquire::Queue::QItem::GetFetchAfter() const /*{{{*/
+{
+   time_point FetchAfter{};
+   for (auto const &O : Owners)
+      FetchAfter = std::max(FetchAfter, O->FetchAfter());
+
+   return FetchAfter;
+}
+									/*}}}*/
 void pkgAcquire::Queue::QItem::SyncDestinationFiles() const		/*{{{*/
 {
    /* ensure that the first owner has the best partial file of all and
@@ -1294,10 +1351,7 @@ pkgAcquireStatus::pkgAcquireStatus() : d(NULL), Percent(-1), Update(true), MoreP
    as well as the current CPS estimate. */
 static struct timeval GetTimevalFromSteadyClock()
 {
-   auto const Time = std::chrono::steady_clock::now().time_since_epoch();
-   auto const Time_sec = std::chrono::duration_cast<std::chrono::seconds>(Time);
-   auto const Time_usec = std::chrono::duration_cast<std::chrono::microseconds>(Time - Time_sec);
-   return { Time_sec.count(), Time_usec.count() };
+   return SteadyDurationToTimeVal(std::chrono::steady_clock::now().time_since_epoch());
 }
 bool pkgAcquireStatus::Pulse(pkgAcquire *Owner)
 {
diff --git a/apt-pkg/acquire.h b/apt-pkg/acquire.h
index a2c4fbc67..09bb408bc 100644
--- a/apt-pkg/acquire.h
+++ b/apt-pkg/acquire.h
@@ -69,6 +69,7 @@
 #include <apt-pkg/macros.h>
 #include <apt-pkg/weakptr.h>
 
+#include <chrono>
 #include <string>
 #include <vector>
 
@@ -92,6 +93,10 @@ class metaIndex;
 class APT_PUBLIC pkgAcquire
 {   
    private:
+   /** \brief The monotonic clock used by the Acquire system */
+   using clock = std::chrono::steady_clock;
+   /** \brief Time point on our monotonic clock */
+   using time_point = std::chrono::time_point<clock>;
    /** \brief FD of the Lock file we acquire in Setup (if any) */
    int LockFD;
    /** \brief dpointer placeholder (for later in case we need it) */
@@ -458,6 +463,8 @@ class APT_PUBLIC pkgAcquire::Queue
       std::string Custom600Headers() const;
       /** @return the maximum priority of this item */
       int APT_HIDDEN GetPriority() const;
+      /** @return the maximum time to fetch this item at */
+      time_point APT_HIDDEN GetFetchAfter() const;
    };
 
    /** \brief The name of this queue. */