src/adaptation/Iterator.cc

   1 /*
   2  * Copyright (C) 1996-2020 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 /* DEBUG: section 93    Adaptation */
  10
  11 #include "squid.h"
  12 #include "adaptation/Answer.h"
  13 #include "adaptation/Config.h"
  14 #include "adaptation/Iterator.h"
  15 #include "adaptation/Service.h"
  16 #include "adaptation/ServiceFilter.h"
  17 #include "adaptation/ServiceGroups.h"
  18 #include "base/TextException.h"
  19 #include "HttpReply.h"
  20 #include "sbuf/StringConvert.h"
  21
  22 Adaptation::Iterator::Iterator(
  23     Http::Message *aMsg, HttpRequest *aCause,
  24     AccessLogEntry::Pointer &alp,
  25     const ServiceGroupPointer &aGroup):
  26     AsyncJob("Iterator"),
  27     Adaptation::Initiate("Iterator"),
  28     theGroup(aGroup),
  29     theMsg(aMsg),
  30     theCause(aCause),
  31     al(alp),
  32     theLauncher(0),
  33     iterations(0),
  34     adapted(false)
  35 {
  36     if (theCause != NULL)
  37         HTTPMSGLOCK(theCause);
  38
  39     if (theMsg != NULL)
  40         HTTPMSGLOCK(theMsg);
  41 }
  42
  43 Adaptation::Iterator::~Iterator()
  44 {
  45     assert(!theLauncher);
  46     HTTPMSGUNLOCK(theMsg);
  47     HTTPMSGUNLOCK(theCause);
  48 }
  49
  50 void Adaptation::Iterator::start()
  51 {
  52     Adaptation::Initiate::start();
  53
  54     thePlan = ServicePlan(theGroup, filter());
  55
  56     // Add adaptation group name once and now, before
  57     // dynamic groups change it at step() time.
  58     if (Adaptation::Config::needHistory && !thePlan.exhausted() && (dynamic_cast<ServiceSet *>(theGroup.getRaw()) || dynamic_cast<ServiceChain *>(theGroup.getRaw()))) {
  59         HttpRequest *request = dynamic_cast<HttpRequest*>(theMsg);
  60         if (!request)
  61             request = theCause;
  62         Must(request);
  63         Adaptation::History::Pointer ah = request->adaptHistory(true);
  64         auto gid = StringToSBuf(theGroup->id);
  65         ah->recordAdaptationService(gid);
  66     }
  67
  68     step();
  69 }
  70
  71 void Adaptation::Iterator::step()
  72 {
  73     ++iterations;
  74     debugs(93,5, HERE << '#' << iterations << " plan: " << thePlan);
  75
  76     Must(!theLauncher);
  77
  78     if (thePlan.exhausted()) { // nothing more to do
  79         sendAnswer(Answer::Forward(theMsg));
  80         Must(done());
  81         return;
  82     }
  83
  84     HttpRequest *request = dynamic_cast<HttpRequest*>(theMsg);
  85     if (!request)
  86         request = theCause;
  87     assert(request);
  88     request->clearError();
  89
  90     if (iterations > Adaptation::Config::service_iteration_limit) {
  91         debugs(93,DBG_CRITICAL, "Adaptation iterations limit (" <<
  92                Adaptation::Config::service_iteration_limit << ") exceeded:\n" <<
  93                "\tPossible service loop with " <<
  94                theGroup->kind << " " << theGroup->id << ", plan=" << thePlan);
  95         throw TexcHere("too many adaptations");
  96     }
  97
  98     ServicePointer service = thePlan.current();
  99     Must(service != NULL);
 100     debugs(93,5, HERE << "using adaptation service: " << service->cfg().key);
 101
 102     if (Adaptation::Config::needHistory) {
 103         Adaptation::History::Pointer ah = request->adaptHistory(true);
 104         auto uid = StringToSBuf(thePlan.current()->cfg().key);
 105         ah->recordAdaptationService(uid);
 106     }
 107
 108     theLauncher = initiateAdaptation(
 109                       service->makeXactLauncher(theMsg, theCause, al));
 110     Must(initiated(theLauncher));
 111     Must(!done());
 112 }
 113
 114 void
 115 Adaptation::Iterator::noteAdaptationAnswer(const Answer &answer)
 116 {
 117     switch (answer.kind) {
 118     case Answer::akForward:
 119         handleAdaptedHeader(const_cast<Http::Message*>(answer.message.getRaw()));
 120         break;
 121
 122     case Answer::akBlock:
 123         handleAdaptationBlock(answer);
 124         break;
 125
 126     case Answer::akError:
 127         handleAdaptationError(answer.final);
 128         break;
 129     }
 130 }
 131
 132 void
 133 Adaptation::Iterator::handleAdaptedHeader(Http::Message *aMsg)
 134 {
 135     // set theCause if we switched to request satisfaction mode
 136     if (!theCause) { // probably sent a request message
 137         if (dynamic_cast<HttpReply*>(aMsg)) { // we got a response message
 138             if (HttpRequest *cause = dynamic_cast<HttpRequest*>(theMsg)) {
 139                 // definately sent request, now use it as the cause
 140                 theCause = cause; // moving the lock
 141                 theMsg = 0;
 142                 debugs(93,3, HERE << "in request satisfaction mode");
 143             }
 144         }
 145     }
 146
 147     Must(aMsg);
 148     HTTPMSGUNLOCK(theMsg);
 149     theMsg = aMsg;
 150     HTTPMSGLOCK(theMsg);
 151     adapted = true;
 152
 153     clearAdaptation(theLauncher);
 154     if (!updatePlan(true)) // do not immediatelly advance the new plan
 155         thePlan.next(filter());
 156     step();
 157 }
 158
 159 void Adaptation::Iterator::noteInitiatorAborted()
 160 {
 161     announceInitiatorAbort(theLauncher); // propogate to the transaction
 162     clearInitiator();
 163     mustStop("initiator gone");
 164 }
 165
 166 void Adaptation::Iterator::handleAdaptationBlock(const Answer &answer)
 167 {
 168     debugs(93,5, HERE << "blocked by " << answer);
 169     clearAdaptation(theLauncher);
 170     updatePlan(false);
 171     sendAnswer(answer);
 172     mustStop("blocked");
 173 }
 174
 175 void Adaptation::Iterator::handleAdaptationError(bool final)
 176 {
 177     debugs(93,5, HERE << "final: " << final << " plan: " << thePlan);
 178     clearAdaptation(theLauncher);
 179     updatePlan(false);
 180
 181     // can we replace the failed service (group-level bypass)?
 182     const bool srcIntact = !theMsg->body_pipe ||
 183                            !theMsg->body_pipe->consumedSize();
 184     // can we ignore the failure (compute while thePlan is not exhausted)?
 185     Must(!thePlan.exhausted());
 186     const bool canIgnore = thePlan.current()->cfg().bypass;
 187     debugs(85,5, HERE << "flags: " << srcIntact << canIgnore << adapted);
 188
 189     if (srcIntact) {
 190         if (thePlan.replacement(filter()) != NULL) {
 191             debugs(93,3, HERE << "trying a replacement service");
 192             step();
 193             return;
 194         }
 195     }
 196
 197     if (canIgnore && srcIntact && adapted) {
 198         debugs(85,3, HERE << "responding with older adapted msg");
 199         sendAnswer(Answer::Forward(theMsg));
 200         mustStop("sent older adapted msg");
 201         return;
 202     }
 203
 204     // caller may recover if we can ignore the error and virgin msg is intact
 205     const bool useVirgin = canIgnore && !adapted && srcIntact;
 206     tellQueryAborted(!useVirgin);
 207     mustStop("group failure");
 208 }
 209
 210 bool Adaptation::Iterator::doneAll() const
 211 {
 212     return Adaptation::Initiate::doneAll() && thePlan.exhausted();
 213 }
 214
 215 void Adaptation::Iterator::swanSong()
 216 {
 217     if (theInitiator.set())
 218         tellQueryAborted(true); // abnormal condition that should not happen
 219
 220     if (initiated(theLauncher))
 221         clearAdaptation(theLauncher);
 222
 223     Adaptation::Initiate::swanSong();
 224 }
 225
 226 bool Adaptation::Iterator::updatePlan(bool adopt)
 227 {
 228     HttpRequest *r = theCause ? theCause : dynamic_cast<HttpRequest*>(theMsg);
 229     Must(r);
 230
 231     Adaptation::History::Pointer ah = r->adaptHistory();
 232     if (!ah) {
 233         debugs(85,9, HERE << "no history to store a service-proposed plan");
 234         return false; // the feature is not enabled or is not triggered
 235     }
 236
 237     String services;
 238     if (!ah->extractNextServices(services)) { // clears history
 239         debugs(85,9, HERE << "no service-proposed plan received");
 240         return false; // the service did not provide a new plan
 241     }
 242
 243     if (!adopt) {
 244         debugs(85,3, HERE << "rejecting service-proposed plan");
 245         return false;
 246     }
 247
 248     debugs(85,3, HERE << "retiring old plan: " << thePlan);
 249
 250     Adaptation::ServiceFilter f = this->filter();
 251     DynamicGroupCfg current, future;
 252     DynamicServiceChain::Split(f, services, current, future);
 253
 254     if (!future.empty()) {
 255         ah->setFutureServices(future);
 256         debugs(85,3, HERE << "noted future service-proposed plan: " << future);
 257     }
 258
 259     // use the current config even if it is empty; we must replace the old plan
 260     theGroup = new DynamicServiceChain(current, f); // refcounted
 261     thePlan = ServicePlan(theGroup, f);
 262     debugs(85,3, HERE << "adopted service-proposed plan: " << thePlan);
 263     return true;
 264 }
 265
 266 Adaptation::ServiceFilter Adaptation::Iterator::filter() const
 267 {
 268     // the method may differ from theGroup->method due to request satisfaction
 269     Method method = methodNone;
 270     // temporary variables, no locking needed
 271     HttpRequest *req = NULL;
 272     HttpReply *rep = NULL;
 273
 274     if (HttpRequest *r = dynamic_cast<HttpRequest*>(theMsg)) {
 275         method = methodReqmod;
 276         req = r;
 277         rep = NULL;
 278     } else if (HttpReply *theReply = dynamic_cast<HttpReply*>(theMsg)) {
 279         method = methodRespmod;
 280         req = theCause;
 281         rep = theReply;
 282     } else {
 283         Must(false); // should not happen
 284     }
 285
 286     return ServiceFilter(method, theGroup->point, req, rep, al);
 287 }
 288
 289 CBDATA_NAMESPACED_CLASS_INIT(Adaptation, Iterator);
 290