src/ipc/Forwarder.cc

   1 /*
   2  * Copyright (C) 1996-2023 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/RequestId.h"
  20 #include "ipc/TypedMsgHdr.h"
  21
  22 Ipc::Forwarder::RequestsMap Ipc::Forwarder::TheRequestsMap;
  23 Ipc::RequestId::Index Ipc::Forwarder::LastRequestId = 0;
  24
  25 Ipc::Forwarder::Forwarder(Request::Pointer aRequest, double aTimeout):
  26     AsyncJob("Ipc::Forwarder"),
  27     codeContext(CodeContext::Current()),
  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, MYNAME);
  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, MYNAME);
  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, MYNAME);
  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, MYNAME);
  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, MYNAME);
 101     Must(param != nullptr);
 102     Forwarder* fwdr = static_cast<Forwarder*>(param);
 103     // use async call to enable job call protection that time events lack
 104
 105     CallBack(fwdr->codeContext, [&fwdr] {
 106         CallJobHere(54, 5, fwdr, Forwarder, requestTimedOut);
 107     });
 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, MYNAME);
 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, 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, 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(const RequestId::Index requestId)
 152 {
 153     debugs(54, 3, MYNAME);
 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 != nullptr);
 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(const RequestId requestId)
 175 {
 176     debugs(54, 3, MYNAME);
 177     Must(requestId != 0);
 178
 179     AsyncCall::Pointer call = DequeueRequest(requestId);
 180     if (call != nullptr)
 181         ScheduleCallHere(call);
 182 }
 183