src/ipc/Strand.cc

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