src/ipc/Strand.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/Subscription.h"
  13 #include "base/TextException.h"
  14 #include "CacheManager.h"
  15 #include "CollapsedForwarding.h"
  16 #include "comm/Connection.h"
  17 #include "fatal.h"
  18 #include "globals.h"
  19 #include "ipc/Kids.h"
  20 #include "ipc/Messages.h"
  21 #include "ipc/SharedListen.h"
  22 #include "ipc/Strand.h"
  23 #include "ipc/StrandCoord.h"
  24 #include "ipc/StrandSearch.h"
  25 #include "mgr/Forwarder.h"
  26 #include "mgr/Request.h"
  27 #include "mgr/Response.h"
  28 #if HAVE_DISKIO_MODULE_IPCIO
  29 #include "DiskIO/IpcIo/IpcIoFile.h" /* XXX: scope boundary violation */
  30 #endif
  31 #if SQUID_SNMP
  32 #include "snmp/Forwarder.h"
  33 #include "snmp/Request.h"
  34 #include "snmp/Response.h"
  35 #endif
  36
  37 CBDATA_NAMESPACED_CLASS_INIT(Ipc, Strand);
  38
  39 Ipc::Strand::Strand():
  40     Port(MakeAddr(strandAddrLabel, KidIdentifier)),
  41     isRegistered(false)
  42 {
  43 }
  44
  45 void Ipc::Strand::start()
  46 {
  47     Port::start();
  48     registerSelf();
  49 }
  50
  51 void Ipc::Strand::registerSelf()
  52 {
  53     debugs(54, 6, HERE);
  54     Must(!isRegistered);
  55
  56     HereIamMessage ann(StrandCoord(KidIdentifier, getpid()));
  57     TypedMsgHdr message;
  58     ann.pack(message);
  59     SendMessage(Port::CoordinatorAddr(), message);
  60     setTimeout(6, "Ipc::Strand::timeoutHandler"); // TODO: make 6 configurable?
  61 }
  62
  63 void Ipc::Strand::receive(const TypedMsgHdr &message)
  64 {
  65     debugs(54, 6, HERE << message.type());
  66     switch (message.type()) {
  67
  68     case mtRegistration:
  69         handleRegistrationResponse(HereIamMessage(message));
  70         break;
  71
  72     case mtSharedListenResponse:
  73         SharedListenJoined(SharedListenResponse(message));
  74         break;
  75
  76 #if HAVE_DISKIO_MODULE_IPCIO
  77     case mtStrandSearchResponse:
  78         IpcIoFile::HandleOpenResponse(StrandSearchResponse(message));
  79         break;
  80
  81     case mtIpcIoNotification:
  82         IpcIoFile::HandleNotification(message);
  83         break;
  84 #endif /* HAVE_DISKIO_MODULE_IPCIO */
  85
  86     case mtCacheMgrRequest: {
  87         const Mgr::Request req(message);
  88         handleCacheMgrRequest(req);
  89     }
  90     break;
  91
  92     case mtCacheMgrResponse: {
  93         const Mgr::Response resp(message);
  94         handleCacheMgrResponse(resp);
  95     }
  96     break;
  97
  98     case mtCollapsedForwardingNotification:
  99         CollapsedForwarding::HandleNotification(message);
 100         break;
 101
 102 #if SQUID_SNMP
 103     case mtSnmpRequest: {
 104         const Snmp::Request req(message);
 105         handleSnmpRequest(req);
 106     }
 107     break;
 108
 109     case mtSnmpResponse: {
 110         const Snmp::Response resp(message);
 111         handleSnmpResponse(resp);
 112     }
 113     break;
 114 #endif
 115
 116     default:
 117         debugs(54, DBG_IMPORTANT, HERE << "Unhandled message type: " << message.type());
 118         break;
 119     }
 120 }
 121
 122 void Ipc::Strand::handleRegistrationResponse(const HereIamMessage &msg)
 123 {
 124     // handle registration response from the coordinator; it could be stale
 125     if (msg.strand.kidId == KidIdentifier && msg.strand.pid == getpid()) {
 126         debugs(54, 6, "kid" << KidIdentifier << " registered");
 127         clearTimeout(); // we are done
 128     } else {
 129         // could be an ACK to the registration message of our dead predecessor
 130         debugs(54, 6, "kid" << KidIdentifier << " is not yet registered");
 131         // keep listening, with a timeout
 132     }
 133 }
 134
 135 void Ipc::Strand::handleCacheMgrRequest(const Mgr::Request& request)
 136 {
 137     Mgr::Action::Pointer action =
 138         CacheManager::GetInstance()->createRequestedAction(request.params);
 139     action->respond(request);
 140 }
 141
 142 void Ipc::Strand::handleCacheMgrResponse(const Mgr::Response& response)
 143 {
 144     Mgr::Forwarder::HandleRemoteAck(response.requestId);
 145 }
 146
 147 #if SQUID_SNMP
 148 void Ipc::Strand::handleSnmpRequest(const Snmp::Request& request)
 149 {
 150     debugs(54, 6, HERE);
 151     Snmp::SendResponse(request.requestId, request.pdu);
 152 }
 153
 154 void Ipc::Strand::handleSnmpResponse(const Snmp::Response& response)
 155 {
 156     debugs(54, 6, HERE);
 157     Snmp::Forwarder::HandleRemoteAck(response.requestId);
 158 }
 159 #endif
 160
 161 void Ipc::Strand::timedout()
 162 {
 163     debugs(54, 6, HERE << isRegistered);
 164     if (!isRegistered)
 165         fatalf("kid%d registration timed out", KidIdentifier);
 166 }
 167