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