src/ipc/SharedListen.cc

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