src/ipc/Forwarder.cc

   1 /*
   2  * $Id$
   3  *
   4  * DEBUG: section 54    Interprocess Communication
   5  *
   6  */
   7
   8 #include "squid.h"
   9 #include "base/AsyncJobCalls.h"
  10 #include "base/TextException.h"
  11 #include "errorpage.h"
  12 #include "HttpReply.h"
  13 #include "HttpRequest.h"
  14 #include "ipc/Forwarder.h"
  15 #include "ipc/Port.h"
  16 #include "ipc/TypedMsgHdr.h"
  17
  18
  19 CBDATA_NAMESPACED_CLASS_INIT(Ipc, Forwarder);
  20
  21 Ipc::Forwarder::RequestsMap Ipc::Forwarder::TheRequestsMap;
  22 unsigned int Ipc::Forwarder::LastRequestId = 0;
  23
  24 Ipc::Forwarder::Forwarder(Request::Pointer aRequest, double aTimeout):
  25         AsyncJob("Ipc::Forwarder"),
  26         request(aRequest), timeout(aTimeout)
  27 {
  28     debugs(54, 5, HERE);
  29 }
  30
  31 Ipc::Forwarder::~Forwarder()
  32 {
  33     debugs(54, 5, HERE);
  34     Must(request->requestId == 0);
  35     cleanup();
  36 }
  37
  38 /// perform cleanup actions
  39 void
  40 Ipc::Forwarder::cleanup()
  41 {
  42 }
  43
  44 void
  45 Ipc::Forwarder::start()
  46 {
  47     debugs(54, 3, HERE);
  48
  49     typedef NullaryMemFunT<Forwarder> Dialer;
  50     AsyncCall::Pointer callback = JobCallback(54, 5, Dialer, this, Forwarder::handleRemoteAck);
  51     if (++LastRequestId == 0) // don't use zero value as request->requestId
  52         ++LastRequestId;
  53     request->requestId = LastRequestId;
  54     TheRequestsMap[request->requestId] = callback;
  55     TypedMsgHdr message;
  56
  57     try {
  58         request->pack(message);
  59     } catch (...) {
  60         // assume the pack() call failed because the message did not fit
  61         // TODO: add a more specific exception?
  62         handleError();
  63     }
  64
  65     SendMessage(coordinatorAddr, message);
  66     eventAdd("Ipc::Forwarder::requestTimedOut", &Forwarder::RequestTimedOut,
  67              this, timeout, 0, false);
  68 }
  69
  70 void
  71 Ipc::Forwarder::swanSong()
  72 {
  73     debugs(54, 5, HERE);
  74     removeTimeoutEvent();
  75     if (request->requestId > 0) {
  76         DequeueRequest(request->requestId);
  77         request->requestId = 0;
  78     }
  79     cleanup();
  80 }
  81
  82 bool
  83 Ipc::Forwarder::doneAll() const
  84 {
  85     debugs(54, 5, HERE);
  86     return request->requestId == 0;
  87 }
  88
  89 /// called when Coordinator starts processing the request
  90 void
  91 Ipc::Forwarder::handleRemoteAck()
  92 {
  93     debugs(54, 3, HERE);
  94     request->requestId = 0;
  95     // Do not clear ENTRY_FWD_HDR_WAIT or do entry->complete() because
  96     // it will trigger our client side processing. Let job cleanup close.
  97 }
  98
  99 /// Ipc::Forwarder::requestTimedOut wrapper
 100 void
 101 Ipc::Forwarder::RequestTimedOut(void* param)
 102 {
 103     debugs(54, 3, HERE);
 104     Must(param != NULL);
 105     Forwarder* fwdr = static_cast<Forwarder*>(param);
 106     // use async call to enable job call protection that time events lack
 107     CallJobHere(54, 5, fwdr, Forwarder, requestTimedOut);
 108 }
 109
 110 /// called when Coordinator fails to start processing the request [in time]
 111 void
 112 Ipc::Forwarder::requestTimedOut()
 113 {
 114     debugs(54, 3, HERE);
 115     handleTimeout();
 116 }
 117
 118 void
 119 Ipc::Forwarder::handleError()
 120 {
 121     mustStop("error");
 122 }
 123
 124 void
 125 Ipc::Forwarder::handleTimeout()
 126 {
 127     mustStop("timeout");
 128 }
 129
 130 /// terminate with an error
 131 void
 132 Ipc::Forwarder::handleException(const std::exception& e)
 133 {
 134     debugs(54, 3, HERE << e.what());
 135     mustStop("exception");
 136 }
 137
 138 void
 139 Ipc::Forwarder::callException(const std::exception& e)
 140 {
 141     try {
 142         handleException(e);
 143     } catch (const std::exception& ex) {
 144         debugs(54, DBG_CRITICAL, HERE << ex.what());
 145     }
 146     AsyncJob::callException(e);
 147 }
 148
 149 /// returns and forgets the right Forwarder callback for the request
 150 AsyncCall::Pointer
 151 Ipc::Forwarder::DequeueRequest(unsigned int requestId)
 152 {
 153     debugs(54, 3, HERE);
 154     Must(requestId != 0);
 155     AsyncCall::Pointer call;
 156     RequestsMap::iterator request = TheRequestsMap.find(requestId);
 157     if (request != TheRequestsMap.end()) {
 158         call = request->second;
 159         Must(call != NULL);
 160         TheRequestsMap.erase(request);
 161     }
 162     return call;
 163 }
 164
 165 /// called when we are no longer waiting for Coordinator to respond
 166 void
 167 Ipc::Forwarder::removeTimeoutEvent()
 168 {
 169     if (eventFind(&Forwarder::RequestTimedOut, this))
 170         eventDelete(&Forwarder::RequestTimedOut, this);
 171 }
 172
 173 void
 174 Ipc::Forwarder::HandleRemoteAck(unsigned int requestId)
 175 {
 176     debugs(54, 3, HERE);
 177     Must(requestId != 0);
 178
 179     AsyncCall::Pointer call = DequeueRequest(requestId);
 180     if (call != NULL)
 181         ScheduleCallHere(call);
 182 }