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 "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(const Comm::ConnectionPointer &c, int anErrNo, int aMapId):
  86         conn(c), errNo(anErrNo), mapId(aMapId)
  87 {
  88 }
  89
  90 Ipc::SharedListenResponse::SharedListenResponse(const TypedMsgHdr &hdrMsg):
  91         conn(NULL), errNo(0), mapId(-1)
  92 {
  93     hdrMsg.checkType(mtSharedListenResponse);
  94     hdrMsg.getPod(*this);
  95     conn = new Comm::Connection;
  96     conn->fd = hdrMsg.getFd();
  97     // other conn details are passed in OpenListenerParams and filled out by SharedListenJoin()
  98 }
  99
 100 void Ipc::SharedListenResponse::pack(TypedMsgHdr &hdrMsg) const
 101 {
 102     hdrMsg.setType(mtSharedListenResponse);
 103     hdrMsg.putPod(*this);
 104     hdrMsg.putFd(conn->fd);
 105 }
 106
 107
 108 void Ipc::JoinSharedListen(const OpenListenerParams &params,
 109                            AsyncCall::Pointer &callback)
 110 {
 111     PendingOpenRequest por;
 112     por.params = params;
 113     por.callback = callback;
 114
 115     SharedListenRequest request;
 116     request.requestorId = KidIdentifier;
 117     request.params = por.params;
 118     request.mapId = AddToMap(por);
 119
 120     debugs(54, 3, HERE << "getting listening FD for " << request.params.addr <<
 121            " mapId=" << request.mapId);
 122
 123     TypedMsgHdr message;
 124     request.pack(message);
 125     SendMessage(coordinatorAddr, message);
 126 }
 127
 128 void Ipc::SharedListenJoined(const SharedListenResponse &response)
 129 {
 130     Comm::ConnectionPointer c = response.conn;
 131
 132     // Dont debugs c fully since only FD is filled right now.
 133     debugs(54, 3, HERE << "got listening FD " << c->fd << " errNo=" <<
 134            response.errNo << " mapId=" << response.mapId);
 135
 136     Must(TheSharedListenRequestMap.find(response.mapId) != TheSharedListenRequestMap.end());
 137     PendingOpenRequest por = TheSharedListenRequestMap[response.mapId];
 138     Must(por.callback != NULL);
 139     TheSharedListenRequestMap.erase(response.mapId);
 140
 141     if (Comm::IsConnOpen(c)) {
 142         OpenListenerParams &p = por.params;
 143         c->local = p.addr;
 144         c->flags = p.flags;
 145         // XXX: leave the comm AI stuff to comm_import_opened()?
 146         struct addrinfo *AI = NULL;
 147         p.addr.GetAddrInfo(AI);
 148         AI->ai_socktype = p.sock_type;
 149         AI->ai_protocol = p.proto;
 150         comm_import_opened(c, FdNote(p.fdNote), AI);
 151         p.addr.FreeAddrInfo(AI);
 152     }
 153
 154     StartListeningCb *cbd = dynamic_cast<StartListeningCb*>(por.callback->getDialer());
 155     Must(cbd);
 156     cbd->conn = c;
 157     cbd->errNo = response.errNo;
 158     cbd->handlerSubscription = por.params.handlerSubscription;
 159     ScheduleCallHere(por.callback);
 160 }