Inquirer.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/Write.h"
15 #include "ipc/Inquirer.h"
16 #include "ipc/Port.h"
17 #include "ipc/TypedMsgHdr.h"
18 #include "MemBuf.h"
19 #include <algorithm>
20 
22 
25 
27 static bool
29 {
30  return c1.kidId < c2.kidId;
31 }
32 
34  double aTimeout):
35  AsyncJob("Ipc::Inquirer"),
36  codeContext(CodeContext::Current()),
37  request(aRequest), strands(coords), pos(strands.begin()), timeout(aTimeout)
38 {
39  debugs(54, 5, MYNAME);
40 
41  // order by ascending kid IDs; useful for non-aggregatable stats
42  std::sort(strands.begin(), strands.end(), LesserStrandByKidId);
43 }
44 
46 {
47  debugs(54, 5, MYNAME);
48  cleanup();
49 }
50 
51 void
53 {
54 }
55 
56 void
58 {
59  request->requestId = 0;
60 }
61 
62 void
64 {
65  if (pos == strands.end()) {
66  Must(done());
67  return;
68  }
69 
70  Must(request->requestId == 0);
71  AsyncCall::Pointer callback = asyncCall(54, 5, "Mgr::Inquirer::handleRemoteAck",
73  if (++LastRequestId == 0) // don't use zero value as request->requestId
74  ++LastRequestId;
75  request->requestId = LastRequestId;
76  const int kidId = pos->kidId;
77  debugs(54, 4, "inquire kid: " << kidId << status());
78  TheRequestsMap[request->requestId] = callback;
79  TypedMsgHdr message;
80  request->pack(message);
81  SendMessage(Port::MakeAddr(strandAddrLabel, kidId), message);
82  eventAdd("Ipc::Inquirer::requestTimedOut", &Inquirer::RequestTimedOut,
83  this, timeout, 0, false);
84 }
85 
87 void
89 {
90  debugs(54, 4, status());
91  request->requestId = 0;
92  removeTimeoutEvent();
93  if (aggregate(response)) {
94  Must(!done()); // or we should not be called
95  ++pos; // advance after a successful inquiry
96  inquire();
97  } else {
98  mustStop("error");
99  }
100 }
101 
102 void
104 {
105  debugs(54, 5, MYNAME);
106  removeTimeoutEvent();
107  if (request->requestId > 0) {
108  DequeueRequest(request->requestId);
109  request->requestId = 0;
110  }
111  sendResponse();
112  cleanup();
113 }
114 
115 bool
117 {
118  return pos == strands.end();
119 }
120 
121 void
122 Ipc::Inquirer::handleException(const std::exception& e)
123 {
124  debugs(54, 3, e.what());
125  mustStop("exception");
126 }
127 
128 void
129 Ipc::Inquirer::callException(const std::exception& e)
130 {
131  debugs(54, 3, MYNAME);
132  try {
133  handleException(e);
134  } catch (const std::exception& ex) {
135  debugs(54, DBG_CRITICAL, ex.what());
136  }
138 }
139 
143 {
144  debugs(54, 3, " requestId " << requestId);
145  Must(requestId != 0);
146  AsyncCall::Pointer call;
147  RequestsMap::iterator request = TheRequestsMap.find(requestId);
148  if (request != TheRequestsMap.end()) {
149  call = request->second;
150  Must(call != NULL);
151  TheRequestsMap.erase(request);
152  }
153  return call;
154 }
155 
156 void
158 {
159  Must(response.requestId != 0);
160  AsyncCall::Pointer call = DequeueRequest(response.requestId);
161  if (call != NULL) {
162  HandleAckDialer* dialer = dynamic_cast<HandleAckDialer*>(call->getDialer());
163  Must(dialer);
164  dialer->arg1 = response.clone();
165  ScheduleCallHere(call);
166  }
167 }
168 
170 void
172 {
175 }
176 
178 void
180 {
181  debugs(54, 3, MYNAME);
182  Must(param != NULL);
183  Inquirer* cmi = static_cast<Inquirer*>(param);
184  // use async call to enable job call protection that time events lack
185  CallBack(cmi->codeContext, [&cmi] {
186  CallJobHere(54, 5, cmi, Inquirer, requestTimedOut);
187  });
188 }
189 
191 void
193 {
194  debugs(54, 3, MYNAME);
195  if (request->requestId != 0) {
196  DequeueRequest(request->requestId);
197  request->requestId = 0;
198  Must(!done()); // or we should not be called
199  ++pos; // advance after a failed inquiry
200  inquire();
201  }
202 }
203 
204 const char*
206 {
207  static MemBuf buf;
208  buf.reset();
209  buf.appendf(" [requestId %u]", request->requestId.index());
210  buf.terminate();
211  return buf.content();
212 }
213 
#define ScheduleCallHere(call)
Definition: AsyncCall.h:164
AsyncCall * asyncCall(int aDebugSection, int aDebugLevel, const char *aName, const Dialer &aDialer)
Definition: AsyncCall.h:154
void CallBack(const CodeContext::Pointer &callbackContext, Fun &&callback)
Definition: CodeContext.h:112
#define Must(condition)
Definition: TextException.h:71
virtual CallDialer * getDialer()=0
virtual void callException(const std::exception &e)
called when the job throws during an async call
Definition: AsyncJob.cc:128
virtual bool doneAll() const
whether positive goal has been reached
Definition: Inquirer.cc:116
static RequestId::Index LastRequestId
last requestId used
Definition: Inquirer.h:85
static void RequestTimedOut(void *param)
Ipc::Inquirer::requestTimedOut wrapper.
Definition: Inquirer.cc:179
virtual void swanSong()
Definition: Inquirer.cc:103
virtual void handleException(const std::exception &e)
do specific exception handling
Definition: Inquirer.cc:122
static RequestsMap TheRequestsMap
pending strand requests
Definition: Inquirer.h:83
CodeContextPointer codeContext
Definition: Inquirer.h:42
virtual void callException(const std::exception &e)
called when the job throws during an async call
Definition: Inquirer.cc:129
static AsyncCall::Pointer DequeueRequest(RequestId::Index)
returns and forgets the right Inquirer callback for strand request
Definition: Inquirer.cc:142
virtual void cleanup()
perform cleanup actions on completion of job
Definition: Inquirer.cc:52
virtual void start()
called by AsyncStart; do not call directly
Definition: Inquirer.cc:57
virtual void inquire()
inquire the next strand
Definition: Inquirer.cc:63
std::map< RequestId::Index, AsyncCall::Pointer > RequestsMap
maps request->id to Inquirer::handleRemoteAck callback
Definition: Inquirer.h:82
Inquirer(Request::Pointer aRequest, const Ipc::StrandCoords &coords, double aTimeout)
Definition: Inquirer.cc:33
virtual const char * status() const
internal cleanup; do not call directly
Definition: Inquirer.cc:205
static void HandleRemoteAck(const Response &response)
finds and calls the right Inquirer upon strand's response
Definition: Inquirer.cc:157
Ipc::StrandCoords strands
all strands we want to query, in order
Definition: Inquirer.h:76
void removeTimeoutEvent()
called when we are no longer waiting for the strand to respond
Definition: Inquirer.cc:171
virtual ~Inquirer()
Definition: Inquirer.cc:45
void requestTimedOut()
called when the strand failed to respond (or finish responding) in time
Definition: Inquirer.cc:192
void handleRemoteAck(Response::Pointer response)
called when a strand is done writing its output
Definition: Inquirer.cc:88
static String MakeAddr(const char *proccessLabel, int id)
calculates IPC message address for strand id of processLabel type
Definition: Port.cc:52
unsigned int Index
Definition: RequestId.h:27
A response to Ipc::Request.
Definition: Response.h:24
RequestId requestId
the ID of the request we are responding to
Definition: Response.h:36
virtual Pointer clone() const =0
returns a copy of this
Strand location details.
Definition: StrandCoord.h:22
int kidId
internal Squid process number
Definition: StrandCoord.h:31
struct msghdr with a known type, fixed-size I/O and control buffers
Definition: TypedMsgHdr.h:33
Definition: MemBuf.h:24
char * content()
start of the added data
Definition: MemBuf.h:41
void reset()
Definition: MemBuf.cc:129
void terminate()
Definition: MemBuf.cc:241
void appendf(const char *fmt,...) PRINTF_FORMAT_ARG2
Append operation with printf-style arguments.
Definition: Packable.h:61
#define MYNAME
Definition: Stream.h:238
#define debugs(SECTION, LEVEL, CONTENT)
Definition: Stream.h:196
#define DBG_CRITICAL
Definition: Stream.h:40
int eventFind(EVH *func, void *arg)
Definition: event.cc:153
void eventDelete(EVH *func, void *arg)
Definition: event.cc:129
void eventAdd(const char *name, EVH *func, void *arg, double when, int weight, bool cbdata)
Definition: event.cc:107
CBDATA_NAMESPACED_CLASS_INIT(Ipc, Inquirer)
static bool LesserStrandByKidId(const Ipc::StrandCoord &c1, const Ipc::StrandCoord &c2)
compare Ipc::StrandCoord using kidId, for std::sort() below
Definition: Inquirer.cc:28
Definition: IpcIoFile.h:24
void SendMessage(const String &toAddress, const TypedMsgHdr &message)
Definition: UdsOp.cc:188
const char strandAddrLabel[]
strand's listening address unique label
Definition: Port.cc:23
std::vector< StrandCoord > StrandCoords
a collection of strand coordinates; the order, if any, is owner-dependent
Definition: StrandCoords.h:19
struct _request * request(char *urlin)
Definition: tcp-banger2.c:291
#define NULL
Definition: types.h:166

 

Introduction

Documentation

Support

Miscellaneous

Web Site Translations

Mirrors