src/ipc/Forwarder.cc

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