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