src/ipc/SharedListen.cc

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