]>
git.ipfire.org Git - thirdparty/squid.git/blob - src/ipc/Strand.cc
2 * DEBUG: section 54 Interprocess Communication
7 #include "base/Subscription.h"
8 #include "base/TextException.h"
9 #include "comm/Connection.h"
11 #include "ipc/Strand.h"
12 #include "ipc/StrandCoord.h"
13 #include "ipc/Messages.h"
14 #include "ipc/SharedListen.h"
15 #include "ipc/StrandSearch.h"
17 #include "mgr/Request.h"
18 #include "mgr/Response.h"
19 #include "mgr/Forwarder.h"
20 #include "SwapDir.h" /* XXX: scope boundary violation */
21 #include "CacheManager.h"
23 #include "DiskIO/IpcIo/IpcIoFile.h" /* XXX: scope boundary violation */
26 #include "snmp/Forwarder.h"
27 #include "snmp/Request.h"
28 #include "snmp/Response.h"
31 CBDATA_NAMESPACED_CLASS_INIT(Ipc
, Strand
);
33 Ipc::Strand::Strand():
34 Port(MakeAddr(strandAddrPfx
, KidIdentifier
)),
39 void Ipc::Strand::start()
45 void Ipc::Strand::registerSelf()
50 HereIamMessage
ann(StrandCoord(KidIdentifier
, getpid()));
53 SendMessage(coordinatorAddr
, message
);
54 setTimeout(6, "Ipc::Strand::timeoutHandler"); // TODO: make 6 configurable?
57 void Ipc::Strand::receive(const TypedMsgHdr
&message
)
59 debugs(54, 6, HERE
<< message
.type());
60 switch (message
.type()) {
63 handleRegistrationResponse(HereIamMessage(message
));
66 case mtSharedListenResponse
:
67 SharedListenJoined(SharedListenResponse(message
));
71 case mtStrandSearchResponse
:
72 IpcIoFile::HandleOpenResponse(StrandSearchResponse(message
));
75 case mtIpcIoNotification
:
76 IpcIoFile::HandleNotification(message
);
78 #endif /* USE_DISKIO_IPCIO */
80 case mtCacheMgrRequest
: {
81 const Mgr::Request
req(message
);
82 handleCacheMgrRequest(req
);
86 case mtCacheMgrResponse
: {
87 const Mgr::Response
resp(message
);
88 handleCacheMgrResponse(resp
);
94 const Snmp::Request
req(message
);
95 handleSnmpRequest(req
);
99 case mtSnmpResponse
: {
100 const Snmp::Response
resp(message
);
101 handleSnmpResponse(resp
);
107 debugs(54, DBG_IMPORTANT
, HERE
<< "Unhandled message type: " << message
.type());
112 void Ipc::Strand::handleRegistrationResponse(const HereIamMessage
&msg
)
114 // handle registration response from the coordinator; it could be stale
115 if (msg
.strand
.kidId
== KidIdentifier
&& msg
.strand
.pid
== getpid()) {
116 debugs(54, 6, "kid" << KidIdentifier
<< " registered");
117 clearTimeout(); // we are done
119 // could be an ACK to the registration message of our dead predecessor
120 debugs(54, 6, "kid" << KidIdentifier
<< " is not yet registered");
121 // keep listening, with a timeout
125 void Ipc::Strand::handleCacheMgrRequest(const Mgr::Request
& request
)
127 Mgr::Action::Pointer action
=
128 CacheManager::GetInstance()->createRequestedAction(request
.params
);
129 action
->respond(request
);
132 void Ipc::Strand::handleCacheMgrResponse(const Mgr::Response
& response
)
134 Mgr::Forwarder::HandleRemoteAck(response
.requestId
);
138 void Ipc::Strand::handleSnmpRequest(const Snmp::Request
& request
)
141 Snmp::SendResponse(request
.requestId
, request
.pdu
);
144 void Ipc::Strand::handleSnmpResponse(const Snmp::Response
& response
)
147 Snmp::Forwarder::HandleRemoteAck(response
.requestId
);
151 void Ipc::Strand::timedout()
153 debugs(54, 6, HERE
<< isRegistered
);
155 fatalf("kid%d registration timed out", KidIdentifier
);