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