SharedListen.cc
Go to the documentation of this file.
1 /*
2  * Copyright (C) 1996-2022 The Squid Software Foundation and contributors
3  *
4  * Squid software is distributed under GPLv2+ license and includes
5  * contributions from numerous individuals and organizations.
6  * Please see the COPYING and CONTRIBUTORS files for details.
7  */
8 
9 /* DEBUG: section 54 Interprocess Communication */
10 
11 #include "squid.h"
12 #include "base/TextException.h"
13 #include "comm.h"
14 #include "comm/Connection.h"
15 #include "globals.h"
16 #include "ipc/Kids.h"
17 #include "ipc/Messages.h"
18 #include "ipc/Port.h"
19 #include "ipc/SharedListen.h"
20 #include "ipc/StartListening.h"
21 #include "ipc/TypedMsgHdr.h"
22 #include "tools.h"
23 
24 #include <list>
25 #include <map>
26 
29 {
30 public:
32  AsyncCall::Pointer callback; // who to notify
33 };
34 
36 typedef std::map<Ipc::RequestId::Index, PendingOpenRequest> SharedListenRequestMap;
38 
40 typedef std::list<PendingOpenRequest> DelayedSharedListenRequests;
42 
43 // TODO: Encapsulate "Pending Request Map" logic shared by all RequestId users.
48 {
49  static Ipc::RequestId::Index LastIndex = 0;
50  // TODO: Switch Ipc::RequestId::Index to uint64_t and drop these 0 checks.
51  if (++LastIndex == 0) // don't use zero value as an ID
52  ++LastIndex;
54  TheSharedListenRequestMap[LastIndex] = por;
55  return LastIndex;
56 }
57 
58 bool
60 {
61  if (sock_type != p.sock_type)
62  return sock_type < p.sock_type;
63 
64  if (proto != p.proto)
65  return proto < p.proto;
66 
67  // ignore flags and fdNote differences because they do not affect binding
68 
69  return addr.compareWhole(p.addr) < 0;
70 }
71 
73  requestorId(KidIdentifier),
74  params(aParams),
75  mapId(aMapId)
76 {
77  // caller will then set public data members
78 }
79 
81 {
83  // XXX: our handlerSubscription is not a POD!
84  hdrMsg.getPod(*this);
85 }
86 
88 {
90  // XXX: our handlerSubscription is not a POD!
91  hdrMsg.putPod(*this);
92 }
93 
94 Ipc::SharedListenResponse::SharedListenResponse(const int aFd, const int anErrNo, const RequestId aMapId):
95  fd(aFd), errNo(anErrNo), mapId(aMapId)
96 {
97 }
98 
100  fd(-1),
101  errNo(0)
102 {
104  hdrMsg.getPod(*this);
105  fd = hdrMsg.getFd();
106  // other conn details are passed in OpenListenerParams and filled out by SharedListenJoin()
107 }
108 
110 {
112  hdrMsg.putPod(*this);
113  // XXX: When we respond with an error, putFd() throws due to the negative fd
114  hdrMsg.putFd(fd);
115 }
116 
117 static void
119 {
121 
122  debugs(54, 3, "getting listening FD for " << request.params.addr <<
123  " mapId=" << request.mapId);
124 
125  Ipc::TypedMsgHdr message;
126  request.pack(message);
128 }
129 
130 static void
132 {
133  if (TheDelayedRequests.empty())
134  return; // no pending requests to resume
135 
136  debugs(54, 3, "resuming with " << TheSharedListenRequestMap.size() <<
137  " active + " << TheDelayedRequests.size() << " delayed requests");
138 
140  TheDelayedRequests.pop_front();
141 }
142 
143 void
145 {
146  PendingOpenRequest por;
147  por.params = params;
148  por.callback = cb;
149 
150  const DelayedSharedListenRequests::size_type concurrencyLimit = 1;
151  if (TheSharedListenRequestMap.size() >= concurrencyLimit) {
152  debugs(54, 3, "waiting for " << TheSharedListenRequestMap.size() <<
153  " active + " << TheDelayedRequests.size() << " delayed requests");
154  TheDelayedRequests.push_back(por);
155  } else {
157  }
158 }
159 
161 {
162  // Dont debugs c fully since only FD is filled right now.
163  debugs(54, 3, "got listening FD " << response.fd << " errNo=" <<
164  response.errNo << " mapId=" << response.mapId << " with " <<
165  TheSharedListenRequestMap.size() << " active + " <<
166  TheDelayedRequests.size() << " delayed requests");
167 
168  Must(response.mapId);
169  const auto pori = TheSharedListenRequestMap.find(response.mapId.index());
170  Must(pori != TheSharedListenRequestMap.end());
171  auto por = pori->second;
172  Must(por.callback);
173  TheSharedListenRequestMap.erase(pori);
174 
175  StartListeningCb *cbd = dynamic_cast<StartListeningCb*>(por.callback->getDialer());
176  assert(cbd && cbd->conn != NULL);
177  Must(cbd && cbd->conn != NULL);
178  cbd->conn->fd = response.fd;
179 
180  if (Comm::IsConnOpen(cbd->conn)) {
181  OpenListenerParams &p = por.params;
182  cbd->conn->local = p.addr;
183  cbd->conn->flags = p.flags;
184  // XXX: leave the comm AI stuff to comm_import_opened()?
185  struct addrinfo *AI = NULL;
186  p.addr.getAddrInfo(AI);
187  AI->ai_socktype = p.sock_type;
188  AI->ai_protocol = p.proto;
189  comm_import_opened(cbd->conn, FdNote(p.fdNote), AI);
191  }
192 
193  cbd->errNo = response.errNo;
194  cbd->handlerSubscription = por.params.handlerSubscription;
195  ScheduleCallHere(por.callback);
196 
198 }
199 
int fd
opened listening socket or -1
Definition: SharedListen.h:74
int errNo
errno value from the comm_open_listener() call
SharedListenRequest(const OpenListenerParams &, RequestId aMapId)
sender's constructor
Definition: SharedListen.cc:72
void setType(int aType)
sets message type; use MessageType enum
Definition: TypedMsgHdr.cc:100
void checkType(int aType) const
Definition: TypedMsgHdr.cc:94
int KidIdentifier
@ mtSharedListenResponse
Definition: Messages.h:29
#define ScheduleCallHere(call)
Definition: AsyncCall.h:164
a request for a listen socket with given parameters
Definition: SharedListen.h:48
int fdNote
index into fd_note() comment strings
Definition: SharedListen.h:34
static void FreeAddr(struct addrinfo *&ai)
Definition: Address.cc:686
void comm_import_opened(const Comm::ConnectionPointer &conn, const char *note, struct addrinfo *AI)
update Comm state after getting a comm_open() FD from another process
Definition: comm.cc:510
int compareWhole(const Ip::Address &rhs) const
Definition: Address.cc:724
bool operator<(const OpenListenerParams &p) const
useful for map<>
Definition: SharedListen.cc:59
bool IsConnOpen(const Comm::ConnectionPointer &conn)
Definition: Connection.cc:27
const char * FdNote(int fdNodeId)
converts FdNoteId into a string
Definition: FdNotes.cc:16
"shared listen" is when concurrent processes are listening on the same fd
Definition: SharedListen.h:27
@ mtSharedListenRequest
Definition: Messages.h:28
void JoinSharedListen(const OpenListenerParams &, AsyncCall::Pointer &)
prepare and send SharedListenRequest to Coordinator
int errNo
errno value from comm_open_sharedListen() call
Definition: SharedListen.h:75
void putPod(const Pod &pod)
store POD
Definition: TypedMsgHdr.h:127
static String CoordinatorAddr()
get the IPC message address for coordinator process
Definition: Port.cc:65
Ipc::OpenListenerParams params
actual comm_open_sharedListen() parameters
Definition: SharedListen.cc:31
#define NULL
Definition: types.h:166
Ip::Address local
Definition: Connection.h:144
void SendMessage(const String &toAddress, const TypedMsgHdr &message)
Definition: UdsOp.cc:188
#define assert(EX)
Definition: assert.h:19
int getFd() const
returns stored descriptor
Definition: TypedMsgHdr.cc:217
SharedListenResponse(int fd, int errNo, RequestId aMapId)
sender's constructor
Definition: SharedListen.cc:94
void pack(TypedMsgHdr &hdrMsg) const
prepare for sendmsg()
static SharedListenRequestMap TheSharedListenRequestMap
Definition: SharedListen.cc:37
holds information necessary to handle JoinListen response
Definition: SharedListen.cc:29
RequestId mapId
to map future response to the requestor's callback
Definition: SharedListen.h:76
static Ipc::RequestId::Index AddToMap(const PendingOpenRequest &por)
Definition: SharedListen.cc:47
static void kickDelayedRequest()
void getAddrInfo(struct addrinfo *&ai, int force=AF_UNSPEC) const
Definition: Address.cc:599
a response to SharedListenRequest
Definition: SharedListen.h:64
Index index() const
Definition: RequestId.h:45
Ip::Address addr
will be memset and memcopied
Definition: SharedListen.h:37
Subscription::Pointer handlerSubscription
The subscription we will pass on to the ConnAcceptor.
#define Must(condition)
Definition: TextException.h:71
static DelayedSharedListenRequests TheDelayedRequests
Definition: SharedListen.cc:41
unsigned int Index
Definition: RequestId.h:27
struct msghdr with a known type, fixed-size I/O and control buffers
Definition: TypedMsgHdr.h:33
void SharedListenJoined(const SharedListenResponse &response)
process Coordinator response to SharedListenRequest
std::map< Ipc::RequestId::Index, PendingOpenRequest > SharedListenRequestMap
maps ID assigned at request time to the response callback
Definition: SharedListen.cc:36
std::list< PendingOpenRequest > DelayedSharedListenRequests
accumulates delayed requests until they are ready to be sent, in FIFO order
Definition: SharedListen.cc:40
void putFd(int aFd)
stores descriptor
Definition: TypedMsgHdr.cc:196
common API for all StartListening() callbacks
void getPod(Pod &pod) const
load POD
Definition: TypedMsgHdr.h:116
AsyncCall::Pointer callback
Definition: SharedListen.cc:32
struct _request * request(char *urlin)
Definition: tcp-banger2.c:291
#define debugs(SECTION, LEVEL, CONTENT)
Definition: Stream.h:196
static void SendSharedListenRequest(const PendingOpenRequest &por)
Comm::ConnectionPointer conn
opened listening socket
void pack(TypedMsgHdr &hdrMsg) const
prepare for sendmsg()
Definition: SharedListen.cc:87

 

Introduction

Documentation

Support

Miscellaneous

Web Site Translations

Mirrors