src/ipc/Coordinator.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 "CacheManager.h"
  12 #include "comm.h"
  13 #include "comm/Connection.h"
  14 #include "ipc/Coordinator.h"
  15 #include "ipc/SharedListen.h"
  16 #include "mgr/Inquirer.h"
  17 #include "mgr/Request.h"
  18 #include "mgr/Response.h"
  19 #include "protos.h"
  20 #if SQUID_SNMP
  21 #include "snmp/Inquirer.h"
  22 #include "snmp/Request.h"
  23 #include "snmp/Response.h"
  24 #endif
  25 #if HAVE_ERRNO_H
  26 #include <errno.h>
  27 #endif
  28
  29 CBDATA_NAMESPACED_CLASS_INIT(Ipc, Coordinator);
  30 Ipc::Coordinator* Ipc::Coordinator::TheInstance = NULL;
  31
  32 Ipc::Coordinator::Coordinator():
  33         Port(coordinatorAddr)
  34 {
  35 }
  36
  37 void Ipc::Coordinator::start()
  38 {
  39     Port::start();
  40 }
  41
  42 Ipc::StrandCoord* Ipc::Coordinator::findStrand(int kidId)
  43 {
  44     typedef StrandCoords::iterator SI;
  45     for (SI iter = strands_.begin(); iter != strands_.end(); ++iter) {
  46         if (iter->kidId == kidId)
  47             return &(*iter);
  48     }
  49     return NULL;
  50 }
  51
  52 void Ipc::Coordinator::registerStrand(const StrandCoord& strand)
  53 {
  54     debugs(54, 3, HERE << "registering kid" << strand.kidId <<
  55            ' ' << strand.tag);
  56     if (StrandCoord* found = findStrand(strand.kidId)) {
  57         const String oldTag = found->tag;
  58         *found = strand;
  59         if (oldTag.size() && !strand.tag.size())
  60             found->tag = oldTag; // keep more detailed info (XXX?)
  61     } else {
  62         strands_.push_back(strand);
  63     }
  64
  65     // notify searchers waiting for this new strand, if any
  66     typedef Searchers::iterator SRI;
  67     for (SRI i = searchers.begin(); i != searchers.end();) {
  68         if (i->tag == strand.tag) {
  69             notifySearcher(*i, strand);
  70             i = searchers.erase(i);
  71         } else {
  72             ++i;
  73         }
  74     }
  75 }
  76
  77 void Ipc::Coordinator::receive(const TypedMsgHdr& message)
  78 {
  79     switch (message.type()) {
  80     case mtRegistration:
  81         debugs(54, 6, HERE << "Registration request");
  82         handleRegistrationRequest(HereIamMessage(message));
  83         break;
  84
  85     case mtStrandSearchRequest: {
  86         const StrandSearchRequest sr(message);
  87         debugs(54, 6, HERE << "Strand search request: " << sr.requestorId <<
  88                " tag: " << sr.tag);
  89         handleSearchRequest(sr);
  90         break;
  91     }
  92
  93     case mtSharedListenRequest:
  94         debugs(54, 6, HERE << "Shared listen request");
  95         handleSharedListenRequest(SharedListenRequest(message));
  96         break;
  97
  98     case mtCacheMgrRequest: {
  99         debugs(54, 6, HERE << "Cache manager request");
 100         const Mgr::Request req(message);
 101         handleCacheMgrRequest(req);
 102     }
 103     break;
 104
 105     case mtCacheMgrResponse: {
 106         debugs(54, 6, HERE << "Cache manager response");
 107         const Mgr::Response resp(message);
 108         handleCacheMgrResponse(resp);
 109     }
 110     break;
 111
 112 #if SQUID_SNMP
 113     case mtSnmpRequest: {
 114         debugs(54, 6, HERE << "SNMP request");
 115         const Snmp::Request req(message);
 116         handleSnmpRequest(req);
 117     }
 118     break;
 119
 120     case mtSnmpResponse: {
 121         debugs(54, 6, HERE << "SNMP response");
 122         const Snmp::Response resp(message);
 123         handleSnmpResponse(resp);
 124     }
 125     break;
 126 #endif
 127
 128     default:
 129         debugs(54, DBG_IMPORTANT, HERE << "Unhandled message type: " << message.type());
 130         break;
 131     }
 132 }
 133
 134 void Ipc::Coordinator::handleRegistrationRequest(const HereIamMessage& msg)
 135 {
 136     registerStrand(msg.strand);
 137
 138     // send back an acknowledgement; TODO: remove as not needed?
 139     TypedMsgHdr message;
 140     msg.pack(message);
 141     SendMessage(MakeAddr(strandAddrPfx, msg.strand.kidId), message);
 142 }
 143
 144 void
 145 Ipc::Coordinator::handleSharedListenRequest(const SharedListenRequest& request)
 146 {
 147     debugs(54, 4, HERE << "kid" << request.requestorId <<
 148            " needs shared listen FD for " << request.params.addr);
 149     Listeners::const_iterator i = listeners.find(request.params);
 150     int errNo = 0;
 151     const Comm::ConnectionPointer c = (i != listeners.end()) ?
 152                                       i->second : openListenSocket(request, errNo);
 153
 154     debugs(54, 3, HERE << "sending shared listen " << c << " for " <<
 155            request.params.addr << " to kid" << request.requestorId <<
 156            " mapId=" << request.mapId);
 157
 158     SharedListenResponse response(c->fd, errNo, request.mapId);
 159     TypedMsgHdr message;
 160     response.pack(message);
 161     SendMessage(MakeAddr(strandAddrPfx, request.requestorId), message);
 162 }
 163
 164 void
 165 Ipc::Coordinator::handleCacheMgrRequest(const Mgr::Request& request)
 166 {
 167     debugs(54, 4, HERE);
 168
 169     try {
 170         Mgr::Action::Pointer action =
 171             CacheManager::GetInstance()->createRequestedAction(request.params);
 172         AsyncJob::Start(new Mgr::Inquirer(action, request, strands_));
 173     } catch (const std::exception &ex) {
 174         debugs(54, DBG_IMPORTANT, "BUG: cannot aggregate mgr:" <<
 175                request.params.actionName << ": " << ex.what());
 176         // TODO: Avoid half-baked Connections or teach them how to close.
 177         ::close(request.conn->fd);
 178         request.conn->fd = -1;
 179         return; // the worker will timeout and close
 180     }
 181
 182     // Let the strand know that we are now responsible for handling the request
 183     Mgr::Response response(request.requestId);
 184     TypedMsgHdr message;
 185     response.pack(message);
 186     SendMessage(MakeAddr(strandAddrPfx, request.requestorId), message);
 187
 188 }
 189
 190 void
 191 Ipc::Coordinator::handleCacheMgrResponse(const Mgr::Response& response)
 192 {
 193     Mgr::Inquirer::HandleRemoteAck(response);
 194 }
 195
 196 void
 197 Ipc::Coordinator::handleSearchRequest(const Ipc::StrandSearchRequest &request)
 198 {
 199     // do we know of a strand with the given search tag?
 200     const StrandCoord *strand = NULL;
 201     typedef StrandCoords::const_iterator SCCI;
 202     for (SCCI i = strands_.begin(); !strand && i != strands_.end(); ++i) {
 203         if (i->tag == request.tag)
 204             strand = &(*i);
 205     }
 206
 207     if (strand) {
 208         notifySearcher(request, *strand);
 209         return;
 210     }
 211
 212     searchers.push_back(request);
 213     debugs(54, 3, HERE << "cannot yet tell kid" << request.requestorId <<
 214            " who " << request.tag << " is");
 215 }
 216
 217 void
 218 Ipc::Coordinator::notifySearcher(const Ipc::StrandSearchRequest &request,
 219                                  const StrandCoord& strand)
 220 {
 221     debugs(54, 3, HERE << "tell kid" << request.requestorId << " that " <<
 222            request.tag << " is kid" << strand.kidId);
 223     const StrandSearchResponse response(strand);
 224     TypedMsgHdr message;
 225     response.pack(message);
 226     SendMessage(MakeAddr(strandAddrPfx, request.requestorId), message);
 227 }
 228
 229 #if SQUID_SNMP
 230 void
 231 Ipc::Coordinator::handleSnmpRequest(const Snmp::Request& request)
 232 {
 233     debugs(54, 4, HERE);
 234
 235     Snmp::Response response(request.requestId);
 236     TypedMsgHdr message;
 237     response.pack(message);
 238     SendMessage(MakeAddr(strandAddrPfx, request.requestorId), message);
 239
 240     AsyncJob::Start(new Snmp::Inquirer(request, strands_));
 241 }
 242
 243 void
 244 Ipc::Coordinator::handleSnmpResponse(const Snmp::Response& response)
 245 {
 246     debugs(54, 4, HERE);
 247     Snmp::Inquirer::HandleRemoteAck(response);
 248 }
 249 #endif
 250
 251 Comm::ConnectionPointer
 252 Ipc::Coordinator::openListenSocket(const SharedListenRequest& request,
 253                                    int &errNo)
 254 {
 255     const OpenListenerParams &p = request.params;
 256
 257     debugs(54, 6, HERE << "opening listen FD at " << p.addr << " for kid" <<
 258            request.requestorId);
 259
 260     Comm::ConnectionPointer conn = new Comm::Connection;
 261     conn->local = p.addr; // comm_open_listener may modify it
 262     conn->flags = p.flags;
 263
 264     enter_suid();
 265     comm_open_listener(p.sock_type, p.proto, conn, FdNote(p.fdNote));
 266     errNo = Comm::IsConnOpen(conn) ? 0 : errno;
 267     leave_suid();
 268
 269     debugs(54, 6, HERE << "tried listening on " << conn << " for kid" <<
 270            request.requestorId);
 271
 272     // cache positive results
 273     if (Comm::IsConnOpen(conn))
 274         listeners[request.params] = conn;
 275
 276     return conn;
 277 }
 278
 279 void Ipc::Coordinator::broadcastSignal(int sig) const
 280 {
 281     typedef StrandCoords::const_iterator SCI;
 282     for (SCI iter = strands_.begin(); iter != strands_.end(); ++iter) {
 283         debugs(54, 5, HERE << "signal " << sig << " to kid" << iter->kidId <<
 284                ", PID=" << iter->pid);
 285         kill(iter->pid, sig);
 286     }
 287 }
 288
 289 Ipc::Coordinator* Ipc::Coordinator::Instance()
 290 {
 291     if (!TheInstance)
 292         TheInstance = new Coordinator;
 293     // XXX: if the Coordinator job quits, this pointer will become invalid
 294     // we could make Coordinator death fatal, except during exit, but since
 295     // Strands do not re-register, even process death would be pointless.
 296     return TheInstance;
 297 }
 298
 299 const Ipc::StrandCoords&
 300 Ipc::Coordinator::strands() const
 301 {
 302     return strands_;
 303 }