src/ipc/SharedListen.cc

   1 /*
   2  * DEBUG: section 54    Interprocess Communication
   3  */
   4
   5 #include "squid.h"
   6 #include "comm.h"
   7 #include "base/TextException.h"
   8 #include "comm/Connection.h"
   9 #include "globals.h"
  10 #include "ipc/Port.h"
  11 #include "ipc/Messages.h"
  12 #include "ipc/Kids.h"
  13 #include "ipc/TypedMsgHdr.h"
  14 #include "ipc/StartListening.h"
  15 #include "ipc/SharedListen.h"
  16 #include "tools.h"
  17
  18 #include <map>
  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 Ipc::SharedListenRequest::SharedListenRequest(): requestorId(-1), mapId(-1)
  66 {
  67     // caller will then set public data members
  68 }
  69
  70 Ipc::SharedListenRequest::SharedListenRequest(const TypedMsgHdr &hdrMsg)
  71 {
  72     hdrMsg.checkType(mtSharedListenRequest);
  73     hdrMsg.getPod(*this);
  74 }
  75
  76 void Ipc::SharedListenRequest::pack(TypedMsgHdr &hdrMsg) const
  77 {
  78     hdrMsg.setType(mtSharedListenRequest);
  79     hdrMsg.putPod(*this);
  80 }
  81
  82 Ipc::SharedListenResponse::SharedListenResponse(int aFd, int anErrNo, int aMapId):
  83         fd(aFd), errNo(anErrNo), mapId(aMapId)
  84 {
  85 }
  86
  87 Ipc::SharedListenResponse::SharedListenResponse(const TypedMsgHdr &hdrMsg):
  88         fd(-1), errNo(0), mapId(-1)
  89 {
  90     hdrMsg.checkType(mtSharedListenResponse);
  91     hdrMsg.getPod(*this);
  92     fd = hdrMsg.getFd();
  93     // other conn details are passed in OpenListenerParams and filled out by SharedListenJoin()
  94 }
  95
  96 void Ipc::SharedListenResponse::pack(TypedMsgHdr &hdrMsg) const
  97 {
  98     hdrMsg.setType(mtSharedListenResponse);
  99     hdrMsg.putPod(*this);
 100     hdrMsg.putFd(fd);
 101 }
 102
 103 void Ipc::JoinSharedListen(const OpenListenerParams &params,
 104                            AsyncCall::Pointer &callback)
 105 {
 106     PendingOpenRequest por;
 107     por.params = params;
 108     por.callback = callback;
 109
 110     SharedListenRequest request;
 111     request.requestorId = KidIdentifier;
 112     request.params = por.params;
 113     request.mapId = AddToMap(por);
 114
 115     debugs(54, 3, HERE << "getting listening FD for " << request.params.addr <<
 116            " mapId=" << request.mapId);
 117
 118     TypedMsgHdr message;
 119     request.pack(message);
 120     SendMessage(coordinatorAddr, message);
 121 }
 122
 123 void Ipc::SharedListenJoined(const SharedListenResponse &response)
 124 {
 125     // Dont debugs c fully since only FD is filled right now.
 126     debugs(54, 3, HERE << "got listening FD " << response.fd << " errNo=" <<
 127            response.errNo << " mapId=" << response.mapId);
 128
 129     Must(TheSharedListenRequestMap.find(response.mapId) != TheSharedListenRequestMap.end());
 130     PendingOpenRequest por = TheSharedListenRequestMap[response.mapId];
 131     Must(por.callback != NULL);
 132     TheSharedListenRequestMap.erase(response.mapId);
 133
 134     StartListeningCb *cbd = dynamic_cast<StartListeningCb*>(por.callback->getDialer());
 135     assert(cbd && cbd->conn != NULL);
 136     Must(cbd && cbd->conn != NULL);
 137     cbd->conn->fd = response.fd;
 138
 139     if (Comm::IsConnOpen(cbd->conn)) {
 140         OpenListenerParams &p = por.params;
 141         cbd->conn->local = p.addr;
 142         cbd->conn->flags = p.flags;
 143         // XXX: leave the comm AI stuff to comm_import_opened()?
 144         struct addrinfo *AI = NULL;
 145         p.addr.GetAddrInfo(AI);
 146         AI->ai_socktype = p.sock_type;
 147         AI->ai_protocol = p.proto;
 148         comm_import_opened(cbd->conn, FdNote(p.fdNote), AI);
 149         p.addr.FreeAddrInfo(AI);
 150     }
 151
 152     cbd->errNo = response.errNo;
 153     cbd->handlerSubscription = por.params.handlerSubscription;
 154     ScheduleCallHere(por.callback);
 155 }