src/adaptation/Iterator.cc

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