src/adaptation/Iterator.cc

   1 /*
   2  * DEBUG: section 93  Adaptation
   3  */
   4
   5 #include "squid.h"
   6 #include "TextException.h"
   7 #include "HttpRequest.h"
   8 #include "HttpReply.h"
   9 #include "HttpMsg.h"
  10 #include "adaptation/Config.h"
  11 #include "adaptation/Iterator.h"
  12 #include "adaptation/Service.h"
  13 #include "adaptation/ServiceFilter.h"
  14 #include "adaptation/ServiceGroups.h"
  15
  16
  17 Adaptation::Iterator::Iterator(Adaptation::Initiator *anInitiator,
  18                                HttpMsg *aMsg, HttpRequest *aCause,
  19                                const ServiceGroupPointer &aGroup):
  20         AsyncJob("Iterator"),
  21         Adaptation::Initiate("Iterator", anInitiator),
  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(theMsg);
  55         Must(done());
  56         return;
  57     }
  58
  59     if (iterations > Adaptation::Config::service_iteration_limit) {
  60         debugs(93,DBG_CRITICAL, "Adaptation iterations limit (" <<
  61                Adaptation::Config::service_iteration_limit << ") exceeded:\n" <<
  62                "\tPossible service loop with " <<
  63                theGroup->kind << " " << theGroup->id << ", plan=" << thePlan);
  64         throw TexcHere("too many adaptations");
  65     }
  66
  67     ServicePointer service = thePlan.current();
  68     Must(service != NULL);
  69     debugs(93,5, HERE << "using adaptation service: " << service->cfg().key);
  70
  71     theLauncher = initiateAdaptation(
  72                       service->makeXactLauncher(this, theMsg, theCause));
  73     Must(theLauncher);
  74     Must(!done());
  75 }
  76
  77 void Adaptation::Iterator::noteAdaptationAnswer(HttpMsg *aMsg)
  78 {
  79     // set theCause if we switched to request satisfaction mode
  80     if (!theCause) { // probably sent a request message
  81         if (dynamic_cast<HttpReply*>(aMsg)) { // we got a response message
  82             if (HttpRequest *cause = dynamic_cast<HttpRequest*>(theMsg)) {
  83                 // definately sent request, now use it as the cause
  84                 theCause = cause; // moving the lock
  85                 theMsg = 0;
  86                 debugs(93,3, HERE << "in request satisfaction mode");
  87             }
  88         }
  89     }
  90
  91     Must(aMsg);
  92     HTTPMSGUNLOCK(theMsg);
  93     theMsg = HTTPMSGLOCK(aMsg);
  94     adapted = true;
  95
  96     clearAdaptation(theLauncher);
  97     if (!updatePlan(true)) // do not immediatelly advance the new plan
  98         thePlan.next(filter());
  99     step();
 100 }
 101
 102 void Adaptation::Iterator::noteInitiatorAborted()
 103 {
 104     announceInitiatorAbort(theLauncher); // propogate to the transaction
 105     clearInitiator();
 106     mustStop("initiator gone");
 107 }
 108
 109 void Adaptation::Iterator::noteAdaptationQueryAbort(bool final)
 110 {
 111     debugs(93,5, HERE << "final: " << final << " plan: " << thePlan);
 112     clearAdaptation(theLauncher);
 113     updatePlan(false);
 114
 115     // can we replace the failed service (group-level bypass)?
 116     const bool srcIntact = !theMsg->body_pipe ||
 117                            !theMsg->body_pipe->consumedSize();
 118     // can we ignore the failure (compute while thePlan is not exhausted)?
 119     Must(!thePlan.exhausted());
 120     const bool canIgnore = thePlan.current()->cfg().bypass;
 121     debugs(85,5, HERE << "flags: " << srcIntact << canIgnore << adapted);
 122
 123     if (srcIntact) {
 124         if (thePlan.replacement(filter()) != NULL) {
 125             debugs(93,3, HERE << "trying a replacement service");
 126             step();
 127             return;
 128         }
 129     }
 130
 131     if (canIgnore && srcIntact && adapted) {
 132         debugs(85,3, HERE << "responding with older adapted msg");
 133         sendAnswer(theMsg);
 134         mustStop("sent older adapted msg");
 135         return;
 136     }
 137
 138     // caller may recover if we can ignore the error and virgin msg is intact
 139     const bool useVirgin = canIgnore && !adapted && srcIntact;
 140     tellQueryAborted(!useVirgin);
 141     mustStop("group failure");
 142 }
 143
 144 bool Adaptation::Iterator::doneAll() const
 145 {
 146     return Adaptation::Initiate::doneAll() && thePlan.exhausted();
 147 }
 148
 149 void Adaptation::Iterator::swanSong()
 150 {
 151     if (theInitiator)
 152         tellQueryAborted(true); // abnormal condition that should not happen
 153
 154     if (theLauncher)
 155         clearAdaptation(theLauncher);
 156
 157     Adaptation::Initiate::swanSong();
 158 }
 159
 160 bool Adaptation::Iterator::updatePlan(bool adopt)
 161 {
 162     HttpRequest *r = theCause ? theCause : dynamic_cast<HttpRequest*>(theMsg);
 163     Must(r);
 164
 165     Adaptation::History::Pointer ah = r->adaptHistory();
 166     if (!ah)
 167         return false; // the feature is not enabled or is not triggered
 168
 169     String services;
 170     if (!ah->extractNextServices(services)) { // clears history
 171         debugs(85,9, HERE << "no service-proposed plan received");
 172         return false; // the service did not provide a new plan
 173     }
 174
 175     if (!adopt) {
 176         debugs(85,3, HERE << "rejecting service-proposed plan");
 177         return false;
 178     }
 179
 180     debugs(85,3, HERE << "retiring old plan: " << thePlan);
 181     theGroup = new DynamicServiceChain(services, theGroup); // refcounted
 182     thePlan = ServicePlan(theGroup, filter());
 183     debugs(85,3, HERE << "adopted service-proposed plan: " << thePlan);
 184     return true;
 185 }
 186
 187 Adaptation::ServiceFilter Adaptation::Iterator::filter() const
 188 {
 189     // the method may differ from theGroup->method due to request satisfaction
 190     Method method = methodNone;
 191     // temporary variables, no locking needed
 192     HttpRequest *req = NULL;
 193     HttpReply *rep = NULL;
 194
 195     if (HttpRequest *r = dynamic_cast<HttpRequest*>(theMsg)) {
 196         method = methodReqmod;
 197         req = r;
 198         rep = NULL;
 199     } else if (HttpReply *theReply = dynamic_cast<HttpReply*>(theMsg)) {
 200         method = methodRespmod;
 201         req = theCause;
 202         rep = theReply;
 203     } else {
 204         Must(false); // should not happen
 205     }
 206
 207     return ServiceFilter(method, theGroup->point, req, rep);
 208 }
 209
 210 CBDATA_NAMESPACED_CLASS_INIT(Adaptation, Iterator);