src/adaptation/ecap/ServiceRep.cc

   1 /*
   2  * DEBUG: section 93    eCAP Interface
   3  */
   4 #include "squid.h"
   5 #include "adaptation/ecap/Config.h"
   6 #include "adaptation/ecap/Host.h"
   7 #include "adaptation/ecap/ServiceRep.h"
   8 #include "adaptation/ecap/XactionRep.h"
   9 #include "AsyncEngine.h"
  10 #include "base/TextException.h"
  11 #include "Debug.h"
  12 #include "EventLoop.h"
  13
  14 #include <libecap/adapter/service.h>
  15 #include <libecap/common/options.h>
  16 #include <libecap/common/name.h>
  17 #include <libecap/common/named_values.h>
  18 #include <limits>
  19 #include <map>
  20
  21 /// libecap::adapter::services indexed by their URI
  22 typedef std::map<std::string, Adaptation::Ecap::ServiceRep::AdapterService> AdapterServices;
  23 /// all loaded services
  24 static AdapterServices TheServices;
  25 /// configured services producing async transactions
  26 static AdapterServices AsyncServices;
  27
  28 namespace Adaptation
  29 {
  30 namespace Ecap
  31 {
  32
  33 /// wraps Adaptation::Ecap::ServiceConfig to allow eCAP visitors
  34 class ConfigRep: public libecap::Options
  35 {
  36 public:
  37     typedef Adaptation::Ecap::ServiceConfig Master;
  38     typedef libecap::Name Name;
  39     typedef libecap::Area Area;
  40
  41     ConfigRep(const Master &aMaster);
  42
  43     // libecap::Options API
  44     virtual const libecap::Area option(const libecap::Name &name) const;
  45     virtual void visitEachOption(libecap::NamedValueVisitor &visitor) const;
  46
  47     const Master &master; ///< the configuration being wrapped
  48 };
  49
  50 /// manages async eCAP transactions
  51 class Engine: public AsyncEngine
  52 {
  53 public:
  54     /* AsyncEngine API */
  55     virtual int checkEvents(int timeout);
  56
  57 private:
  58     void kickAsyncServices(timeval &timeout);
  59 };
  60
  61 } // namespace Ecap
  62 } // namespace Adaptation
  63
  64 Adaptation::Ecap::ConfigRep::ConfigRep(const Master &aMaster): master(aMaster)
  65 {
  66 }
  67
  68 const libecap::Area
  69 Adaptation::Ecap::ConfigRep::option(const libecap::Name &name) const
  70 {
  71     // we may supply the params we know about, but only when names have host ID
  72     if (name == metaBypassable)
  73         return Area(master.bypass ? "1" : "0", 1);
  74
  75     // TODO: We could build a by-name index, but is it worth it? Good adapters
  76     // should use visitEachOption() instead, to check for name typos/errors.
  77     typedef Master::Extensions::const_iterator MECI;
  78     for (MECI i = master.extensions.begin(); i != master.extensions.end(); ++i) {
  79         if (name == i->first)
  80             return Area(i->second.data(), i->second.size());
  81     }
  82
  83     return Area();
  84 }
  85
  86 void
  87 Adaptation::Ecap::ConfigRep::visitEachOption(libecap::NamedValueVisitor &visitor) const
  88 {
  89     // we may supply the params we know about too, but only if we set host ID
  90     visitor.visit(metaBypassable, Area(master.bypass ? "1" : "0", 1));
  91
  92     // visit adapter-specific options (i.e., those not recognized by Squid)
  93     typedef Master::Extensions::const_iterator MECI;
  94     for (MECI i = master.extensions.begin(); i != master.extensions.end(); ++i)
  95         visitor.visit(Name(i->first), Area::FromTempString(i->second));
  96 }
  97
  98 /* Adaptation::Ecap::Engine */
  99
 100 int
 101 Adaptation::Ecap::Engine::checkEvents(int)
 102 {
 103     // Start with the default I/O loop timeout, convert from milliseconds.
 104     static const struct timeval maxTimeout = {
 105         EVENT_LOOP_TIMEOUT/1000, // seconds
 106         (EVENT_LOOP_TIMEOUT % 1000)*1000
 107     }; // microseconds
 108     struct timeval timeout = maxTimeout;
 109
 110     kickAsyncServices(timeout);
 111     if (timeout.tv_sec == maxTimeout.tv_sec && timeout.tv_usec == maxTimeout.tv_usec)
 112         return EVENT_IDLE;
 113
 114     debugs(93, 7, "timeout: " << timeout.tv_sec << "s+" << timeout.tv_usec << "us");
 115
 116     // convert back to milliseconds, avoiding int overflows
 117     if (timeout.tv_sec >= std::numeric_limits<int>::max()/1000 - 1000)
 118         return std::numeric_limits<int>::max();
 119     else
 120         return timeout.tv_sec*1000 + timeout.tv_usec/1000;
 121 }
 122
 123 /// resumes async transactions (if any) and returns true if they set a timeout
 124 void
 125 Adaptation::Ecap::Engine::kickAsyncServices(timeval &timeout)
 126 {
 127     if (AsyncServices.empty())
 128         return;
 129
 130     debugs(93, 3, "async services: " << AsyncServices.size());
 131
 132     // Activate waiting async transactions, if any.
 133     typedef AdapterServices::iterator ASI;
 134     for (ASI s = AsyncServices.begin(); s != AsyncServices.end(); ++s) {
 135         assert(s->second);
 136         s->second->resume(); // may call Ecap::Xaction::resume()
 137     }
 138
 139     // Give services a chance to decrease the default timeout.
 140     for (ASI s = AsyncServices.begin(); s != AsyncServices.end(); ++s) {
 141         s->second->suspend(timeout);
 142     }
 143 }
 144
 145 /* Adaptation::Ecap::ServiceRep */
 146
 147 Adaptation::Ecap::ServiceRep::ServiceRep(const ServiceConfigPointer &cfg):
 148         /*AsyncJob("Adaptation::Ecap::ServiceRep"),*/ Adaptation::Service(cfg),
 149         isDetached(false)
 150 {
 151 }
 152
 153 Adaptation::Ecap::ServiceRep::~ServiceRep()
 154 {
 155 }
 156
 157 void Adaptation::Ecap::ServiceRep::noteFailure()
 158 {
 159     assert(false); // XXX: should this be ICAP-specific?
 160 }
 161
 162 void
 163 Adaptation::Ecap::ServiceRep::finalize()
 164 {
 165     Adaptation::Service::finalize();
 166     theService = FindAdapterService(cfg().uri);
 167     if (theService) {
 168         try {
 169             tryConfigureAndStart();
 170             Must(up());
 171         } catch (const std::exception &e) { // standardized exceptions
 172             if (!handleFinalizeFailure(e.what()))
 173                 throw; // rethrow for upper layers to handle
 174         } catch (...) { // all other exceptions
 175             if (!handleFinalizeFailure("unrecognized exception"))
 176                 throw; // rethrow for upper layers to handle
 177         }
 178         return; // success or handled exception
 179     } else {
 180         debugs(93,DBG_IMPORTANT, "WARNING: configured ecap_service was not loaded: " << cfg().uri);
 181     }
 182 }
 183
 184 /// attempts to configure and start eCAP service; the caller handles exceptions
 185 void
 186 Adaptation::Ecap::ServiceRep::tryConfigureAndStart()
 187 {
 188     debugs(93,2, HERE << "configuring eCAP service: " << theService->uri());
 189     const ConfigRep cfgRep(dynamic_cast<const ServiceConfig&>(cfg()));
 190     theService->configure(cfgRep);
 191
 192     debugs(93,DBG_IMPORTANT, "Starting eCAP service: " << theService->uri());
 193     theService->start();
 194
 195     if (theService->makesAsyncXactions()) {
 196         AsyncServices[theService->uri()] = theService;
 197         debugs(93, 5, "asyncs: " << AsyncServices.size());
 198     }
 199 }
 200
 201 /// handles failures while configuring or starting an eCAP service;
 202 /// returns false if the error must be propagated to higher levels
 203 bool
 204 Adaptation::Ecap::ServiceRep::handleFinalizeFailure(const char *error)
 205 {
 206     const bool salvage = cfg().bypass;
 207     const int level = salvage ? DBG_IMPORTANT :DBG_CRITICAL;
 208     const char *kind = salvage ? "optional" : "essential";
 209     debugs(93, level, "ERROR: failed to start " << kind << " eCAP service: " <<
 210            cfg().uri << ":\n" << error);
 211
 212     if (!salvage)
 213         return false; // we cannot handle the problem; the caller may escalate
 214
 215     // make up() false, preventing new adaptation requests and enabling bypass
 216     theService.reset();
 217     debugs(93, level, "WARNING: " << kind << " eCAP service is " <<
 218            "down after initialization failure: " << cfg().uri);
 219
 220     return true; // tell the caller to ignore the problem because we handled it
 221 }
 222
 223 bool Adaptation::Ecap::ServiceRep::probed() const
 224 {
 225     return true; // we "probe" the adapter in finalize().
 226 }
 227
 228 bool Adaptation::Ecap::ServiceRep::up() const
 229 {
 230     return theService != NULL;
 231 }
 232
 233 bool Adaptation::Ecap::ServiceRep::wantsUrl(const String &urlPath) const
 234 {
 235     Must(up());
 236     return theService->wantsUrl(urlPath.termedBuf());
 237 }
 238
 239 Adaptation::Initiate *
 240 Adaptation::Ecap::ServiceRep::makeXactLauncher(HttpMsg *virgin,
 241         HttpRequest *cause, AccessLogEntry::Pointer &alp)
 242 {
 243     Must(up());
 244
 245     // register now because (a) we need EventLoop::Running and (b) we do not
 246     // want to add more main loop overheads unless an async service is used.
 247     static AsyncEngine *TheEngine = NULL;
 248     if (AsyncServices.size() && !TheEngine && EventLoop::Running) {
 249         TheEngine = new Engine;
 250         EventLoop::Running->registerEngine(TheEngine);
 251         debugs(93, 3, "asyncs: " << AsyncServices.size() << ' ' << TheEngine);
 252     }
 253
 254     XactionRep *rep = new XactionRep(virgin, cause, alp, Pointer(this));
 255     XactionRep::AdapterXaction x(theService->makeXaction(rep));
 256     rep->master(x);
 257     return rep;
 258 }
 259
 260 // returns a temporary string depicting service status, for debugging
 261 const char *Adaptation::Ecap::ServiceRep::status() const
 262 {
 263     // TODO: move generic stuff from eCAP and ICAP to Adaptation
 264     static MemBuf buf;
 265
 266     buf.reset();
 267     buf.append("[", 1);
 268
 269     if (up())
 270         buf.append("up", 2);
 271     else
 272         buf.append("down", 4);
 273
 274     if (detached())
 275         buf.append(",detached", 9);
 276
 277     buf.append("]", 1);
 278     buf.terminate();
 279
 280     return buf.content();
 281 }
 282
 283 void Adaptation::Ecap::ServiceRep::detach()
 284 {
 285     isDetached = true;
 286 }
 287
 288 bool Adaptation::Ecap::ServiceRep::detached() const
 289 {
 290     return isDetached;
 291 }
 292
 293 Adaptation::Ecap::ServiceRep::AdapterService
 294 Adaptation::Ecap::FindAdapterService(const String& serviceUri)
 295 {
 296     AdapterServices::const_iterator pos = TheServices.find(serviceUri.termedBuf());
 297     if (pos != TheServices.end()) {
 298         Must(pos->second);
 299         return pos->second;
 300     }
 301     return ServiceRep::AdapterService();
 302 }
 303
 304 void
 305 Adaptation::Ecap::RegisterAdapterService(const Adaptation::Ecap::ServiceRep::AdapterService& adapterService)
 306 {
 307     TheServices[adapterService->uri()] = adapterService; // may update old one
 308     debugs(93, 3, "stored eCAP module service: " << adapterService->uri());
 309     // We do not update AsyncServices here in case they are not configured.
 310 }
 311
 312 void
 313 Adaptation::Ecap::UnregisterAdapterService(const String& serviceUri)
 314 {
 315     if (TheServices.erase(serviceUri.termedBuf())) {
 316         debugs(93, 3, "unregistered eCAP module service: " << serviceUri);
 317         AsyncServices.erase(serviceUri.termedBuf()); // no-op for non-async
 318         return;
 319     }
 320     debugs(93, 3, "failed to unregister eCAP module service: " << serviceUri);
 321 }
 322
 323 void
 324 Adaptation::Ecap::CheckUnusedAdapterServices(const Adaptation::Services& cfgs)
 325 {
 326     typedef AdapterServices::const_iterator ASCI;
 327     for (ASCI loaded = TheServices.begin(); loaded != TheServices.end();
 328             ++loaded) {
 329         bool found = false;
 330         for (Services::const_iterator cfged = cfgs.begin();
 331                 cfged != cfgs.end() && !found; ++cfged) {
 332             found = (*cfged)->cfg().uri == loaded->second->uri().c_str();
 333         }
 334         if (!found)
 335             debugs(93, DBG_IMPORTANT, "Warning: loaded eCAP service has no matching " <<
 336                    "ecap_service config option: " << loaded->second->uri());
 337     }
 338 }