src/ipc/SharedListen.cc

   1 /*
   2  * $Id$
   3  *
   4  * DEBUG: section 54    Interprocess Communication
   5  *
   6  */
   7
   8 #include "squid.h"
   9 #include "comm.h"
  10 #include "base/TextException.h"
  11 #include "comm/Connection.h"
  12 #include "globals.h"
  13 #include "ipc/Port.h"
  14 #include "ipc/Messages.h"
  15 #include "ipc/Kids.h"
  16 #include "ipc/TypedMsgHdr.h"
  17 #include "ipc/StartListening.h"
  18 #include "ipc/SharedListen.h"
  19 #include "protos.h"
  20
  21 #include <map>
  22
  23 /// holds information necessary to handle JoinListen response
  24 class PendingOpenRequest
  25 {
  26 public:
  27     Ipc::OpenListenerParams params; ///< actual comm_open_sharedListen() parameters
  28     AsyncCall::Pointer callback; // who to notify
  29 };
  30
  31 /// maps ID assigned at request time to the response callback
  32 typedef std::map<int, PendingOpenRequest> SharedListenRequestMap;
  33 static SharedListenRequestMap TheSharedListenRequestMap;
  34
  35 static int
  36 AddToMap(const PendingOpenRequest &por)
  37 {
  38     // find unused ID using linear seach; there should not be many entries
  39     for (int id = 0; true; ++id) {
  40         if (TheSharedListenRequestMap.find(id) == TheSharedListenRequestMap.end()) {
  41             TheSharedListenRequestMap[id] = por;
  42             return id;
  43         }
  44     }
  45     assert(false); // not reached
  46     return -1;
  47 }
  48
  49 Ipc::OpenListenerParams::OpenListenerParams()
  50 {
  51     xmemset(this, 0, sizeof(*this));
  52 }
  53
  54 bool
  55 Ipc::OpenListenerParams::operator <(const OpenListenerParams &p) const
  56 {
  57     if (sock_type != p.sock_type)
  58         return sock_type < p.sock_type;
  59
  60     if (proto != p.proto)
  61         return proto < p.proto;
  62
  63     // ignore flags and fdNote differences because they do not affect binding
  64
  65     return addr.compareWhole(p.addr) < 0;
  66 }
  67
  68 Ipc::SharedListenRequest::SharedListenRequest(): requestorId(-1), mapId(-1)
  69 {
  70     // caller will then set public data members
  71 }
  72
  73 Ipc::SharedListenRequest::SharedListenRequest(const TypedMsgHdr &hdrMsg)
  74 {
  75     hdrMsg.checkType(mtSharedListenRequest);
  76     hdrMsg.getPod(*this);
  77 }
  78
  79 void Ipc::SharedListenRequest::pack(TypedMsgHdr &hdrMsg) const
  80 {
  81     hdrMsg.setType(mtSharedListenRequest);
  82     hdrMsg.putPod(*this);
  83 }
  84
  85 Ipc::SharedListenResponse::SharedListenResponse(int aFd, int anErrNo, int aMapId):
  86         fd(aFd), errNo(anErrNo), mapId(aMapId)
  87 {
  88 }
  89
  90 Ipc::SharedListenResponse::SharedListenResponse(const TypedMsgHdr &hdrMsg):
  91         fd(-1), errNo(0), mapId(-1)
  92 {
  93     hdrMsg.checkType(mtSharedListenResponse);
  94     hdrMsg.getPod(*this);
  95     fd = hdrMsg.getFd();
  96     // other conn details are passed in OpenListenerParams and filled out by SharedListenJoin()
  97 }
  98
  99 void Ipc::SharedListenResponse::pack(TypedMsgHdr &hdrMsg) const
 100 {
 101     hdrMsg.setType(mtSharedListenResponse);
 102     hdrMsg.putPod(*this);
 103     hdrMsg.putFd(fd);
 104 }
 105
 106 void Ipc::JoinSharedListen(const OpenListenerParams &params,
 107                            AsyncCall::Pointer &callback)
 108 {
 109     PendingOpenRequest por;
 110     por.params = params;
 111     por.callback = callback;
 112
 113     SharedListenRequest request;
 114     request.requestorId = KidIdentifier;
 115     request.params = por.params;
 116     request.mapId = AddToMap(por);
 117
 118     debugs(54, 3, HERE << "getting listening FD for " << request.params.addr <<
 119            " mapId=" << request.mapId);
 120
 121     TypedMsgHdr message;
 122     request.pack(message);
 123     SendMessage(coordinatorAddr, message);
 124 }
 125
 126 void Ipc::SharedListenJoined(const SharedListenResponse &response)
 127 {
 128     // Dont debugs c fully since only FD is filled right now.
 129     debugs(54, 3, HERE << "got listening FD " << response.fd << " errNo=" <<
 130            response.errNo << " mapId=" << response.mapId);
 131
 132     Must(TheSharedListenRequestMap.find(response.mapId) != TheSharedListenRequestMap.end());
 133     PendingOpenRequest por = TheSharedListenRequestMap[response.mapId];
 134     Must(por.callback != NULL);
 135     TheSharedListenRequestMap.erase(response.mapId);
 136
 137     StartListeningCb *cbd = dynamic_cast<StartListeningCb*>(por.callback->getDialer());
 138     assert(cbd && cbd->conn != NULL);
 139     Must(cbd && cbd->conn != NULL);
 140     cbd->conn->fd = response.fd;
 141
 142     if (Comm::IsConnOpen(cbd->conn)) {
 143         OpenListenerParams &p = por.params;
 144         cbd->conn->local = p.addr;
 145         cbd->conn->flags = p.flags;
 146         // XXX: leave the comm AI stuff to comm_import_opened()?
 147         struct addrinfo *AI = NULL;
 148         p.addr.GetAddrInfo(AI);
 149         AI->ai_socktype = p.sock_type;
 150         AI->ai_protocol = p.proto;
 151         comm_import_opened(cbd->conn, FdNote(p.fdNote), AI);
 152         p.addr.FreeAddrInfo(AI);
 153     }
 154
 155     cbd->errNo = response.errNo;
 156     cbd->handlerSubscription = por.params.handlerSubscription;
 157     ScheduleCallHere(por.callback);
 158 }