src/ipc/Strand.cc

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