src/adaptation/Iterator.cc

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