1//===------- SimpleRemoteEPC.cpp -- Simple remote executor control --------===//
2//
3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4// See https://llvm.org/LICENSE.txt for license information.
5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6//
7//===----------------------------------------------------------------------===//
8
9#include "llvm/ExecutionEngine/Orc/SimpleRemoteEPC.h"
10#include "llvm/ExecutionEngine/Orc/CallProxiesSPS.h"
11#include "llvm/ExecutionEngine/Orc/EPCGenericDylibManagerSPS.h"
12#include "llvm/ExecutionEngine/Orc/EPCGenericJITLinkMemoryManagerSPS.h"
13#include "llvm/ExecutionEngine/Orc/EPCGenericMemoryAccessSPS.h"
14#include "llvm/ExecutionEngine/Orc/LookupAndApply.h"
15#include "llvm/ExecutionEngine/Orc/RecordProxy.h"
16#include "llvm/ExecutionEngine/Orc/Shared/OrcRTBridge.h"
17#include "llvm/Support/FormatVariadic.h"
18
19#define DEBUG_TYPE "orc"
20
21namespace llvm {
22namespace orc {
23
24SimpleRemoteEPC::~SimpleRemoteEPC() {
25#ifndef NDEBUG
26 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
27 assert(Disconnected && "Destroyed without disconnection");
28#endif // NDEBUG
29}
30
31Expected<int32_t> SimpleRemoteEPC::runAsMain(ExecutorAddr MainFnAddr,
32 ArrayRef<std::string> Args) {
33 if (!CallMain)
34 if (auto Err =
35 lookupAndApply(JD&: getExecutionSession().getBootstrapJITDylib(),
36 PrepareFns: recordProxy<sps::CallMainProxySpec>(P: &CallMain)))
37 return Err;
38
39 auto Result = CallMain(getExecutionSession(), MainFnAddr, Args);
40 if (!Result)
41 return Result.takeError();
42 return *Result;
43}
44
45void SimpleRemoteEPC::callWrapperAsync(ExecutorAddr WrapperFnAddr,
46 IncomingWFRHandler OnComplete,
47 ArrayRef<char> ArgBuffer) {
48 uint64_t SeqNo;
49 {
50 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
51 SeqNo = getNextSeqNo();
52 assert(!PendingCallWrapperResults.count(SeqNo) && "SeqNo already in use");
53 PendingCallWrapperResults[SeqNo] = std::move(OnComplete);
54 }
55
56 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::CallWrapper, SeqNo,
57 TagAddr: WrapperFnAddr, ArgBytes: ArgBuffer)) {
58 IncomingWFRHandler H;
59
60 // We just registered OnComplete, but there may be a race between this
61 // thread returning from sendMessage and handleDisconnect being called from
62 // the transport's listener thread. If handleDisconnect gets there first
63 // then it will have failed 'H' for us. If we get there first (or if
64 // handleDisconnect already ran) then we need to take care of it.
65 {
66 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
67 auto I = PendingCallWrapperResults.find(Val: SeqNo);
68 if (I != PendingCallWrapperResults.end()) {
69 H = std::move(I->second);
70 PendingCallWrapperResults.erase(I);
71 }
72 }
73
74 if (H)
75 H(shared::WrapperFunctionBuffer::createOutOfBandError(Msg: "disconnecting"));
76
77 getExecutionSession().reportError(Err: std::move(Err));
78 }
79}
80
81Expected<std::unique_ptr<jitlink::JITLinkMemoryManager>>
82SimpleRemoteEPC::createDefaultMemoryManager() {
83 return sps::createEPCGenericJITLinkMemoryManager(ES&: getExecutionSession());
84}
85
86Expected<std::unique_ptr<DylibManager>>
87SimpleRemoteEPC::createDefaultDylibMgr() {
88 return sps::createEPCGenericDylibManager(ES&: getExecutionSession());
89}
90
91Expected<std::unique_ptr<MemoryAccess>>
92SimpleRemoteEPC::createDefaultMemoryAccess() {
93 return sps::createEPCGenericMemoryAccess(ES&: getExecutionSession());
94}
95
96Error SimpleRemoteEPC::disconnect() {
97 // disconnect is idempotent, so the first caller owns the hangup. There is
98 // also nothing to announce to an executor that has already announced its own
99 // departure.
100 bool SendHangup = false;
101 {
102 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
103 SendHangup = !LocalHangup && !RemoteHangup;
104 LocalHangup = true;
105 }
106
107 // Tell the executor we're going away, so that it can distinguish this from
108 // losing us unexpectedly. Best-effort: if the send fails there is nothing to
109 // do but tear down anyway, and the executor will report the disconnection as
110 // unexpected. A locally requested disconnect is orderly, so the hangup
111 // carries a success value.
112 if (SendHangup) {
113 auto Payload = encodeHangupPayload(Err: Error::success());
114 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::Hangup, SeqNo: 0, TagAddr: ExecutorAddr(),
115 ArgBytes: {Payload.data(), Payload.size()}))
116 consumeError(Err: std::move(Err));
117 }
118
119 T->disconnect();
120 D->shutdown();
121 std::unique_lock<std::mutex> Lock(SimpleRemoteEPCMutex);
122 DisconnectCV.wait(lock&: Lock, p: [this] { return Disconnected; });
123 return std::move(DisconnectErr);
124}
125
126Expected<SimpleRemoteEPCTransportClient::HandleMessageAction>
127SimpleRemoteEPC::handleMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo,
128 ExecutorAddr TagAddr,
129 shared::WrapperFunctionBuffer ArgBytes) {
130
131 LLVM_DEBUG({
132 dbgs() << "SimpleRemoteEPC::handleMessage: opc = ";
133 switch (OpC) {
134 case SimpleRemoteEPCOpcode::Setup:
135 dbgs() << "Setup";
136 assert(SeqNo == 0 && "Non-zero SeqNo for Setup?");
137 assert(!TagAddr && "Non-zero TagAddr for Setup?");
138 break;
139 case SimpleRemoteEPCOpcode::Hangup:
140 dbgs() << "Hangup";
141 assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
142 assert(!TagAddr && "Non-zero TagAddr for Hangup?");
143 break;
144 case SimpleRemoteEPCOpcode::Result:
145 dbgs() << "Result";
146 assert(!TagAddr && "Non-zero TagAddr for Result?");
147 break;
148 case SimpleRemoteEPCOpcode::CallWrapper:
149 dbgs() << "CallWrapper";
150 break;
151 }
152 dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
153 << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
154 << " bytes\n";
155 });
156
157 using UT = std::underlying_type_t<SimpleRemoteEPCOpcode>;
158 if (static_cast<UT>(OpC) > static_cast<UT>(SimpleRemoteEPCOpcode::LastOpC))
159 return make_error<StringError>(Args: "Unexpected opcode",
160 Args: inconvertibleErrorCode());
161
162 switch (OpC) {
163 case SimpleRemoteEPCOpcode::Setup:
164 if (auto Err = handleSetup(SeqNo, TagAddr, ArgBytes: std::move(ArgBytes)))
165 return std::move(Err);
166 break;
167 case SimpleRemoteEPCOpcode::Hangup:
168 T->disconnect();
169 {
170 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
171 RemoteHangup = true;
172 }
173 if (auto Err = handleHangup(ArgBytes: std::move(ArgBytes)))
174 return std::move(Err);
175 return EndSession;
176 case SimpleRemoteEPCOpcode::Result:
177 if (auto Err = handleResult(SeqNo, TagAddr, ArgBytes: std::move(ArgBytes)))
178 return std::move(Err);
179 break;
180 case SimpleRemoteEPCOpcode::CallWrapper:
181 handleCallWrapper(RemoteSeqNo: SeqNo, TagAddr, ArgBytes: std::move(ArgBytes));
182 break;
183 }
184 return ContinueSession;
185}
186
187void SimpleRemoteEPC::handleDisconnect(Error Err) {
188 LLVM_DEBUG({
189 dbgs() << "SimpleRemoteEPC::handleDisconnect: "
190 << (Err ? "failure" : "success") << "\n";
191 });
192
193 PendingCallWrapperResultsMap TmpPending;
194
195 {
196 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
197 std::swap(a&: TmpPending, b&: PendingCallWrapperResults);
198 }
199
200 for (auto &KV : TmpPending)
201 KV.second(
202 shared::WrapperFunctionBuffer::createOutOfBandError(Msg: "disconnecting"));
203
204 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
205
206 // If the transport reported no error, but neither side announced the end of
207 // the session, then the executor went away without telling us. The cause is
208 // not knowable from here -- it may have crashed, been killed, or become
209 // unreachable -- so report what was observed rather than a cause.
210 //
211 // A missing hangup is evidence, not proof: a hangup can also be lost in
212 // transit, since closing a TCP socket with unread data queued sends an RST,
213 // which can discard bytes the peer had already delivered. We accept that
214 // rather than draining the read side before closing -- the cost is a
215 // misleading diagnostic on a session that is ending regardless, whereas a
216 // drain risks stalling teardown on a peer that never closes.
217 Error DisconnectReason =
218 (!Err && !LocalHangup && !RemoteHangup)
219 ? make_error<StringError>(Args: "Connection closed without hangup",
220 Args: inconvertibleErrorCode())
221 : std::move(Err);
222
223 DisconnectErr =
224 joinErrors(E1: std::move(DisconnectErr), E2: std::move(DisconnectReason));
225 Disconnected = true;
226 DisconnectCV.notify_all();
227}
228
229Error SimpleRemoteEPC::sendMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo,
230 ExecutorAddr TagAddr,
231 ArrayRef<char> ArgBytes) {
232 assert(OpC != SimpleRemoteEPCOpcode::Setup &&
233 "SimpleRemoteEPC sending Setup message? That's the wrong direction.");
234
235 LLVM_DEBUG({
236 dbgs() << "SimpleRemoteEPC::sendMessage: opc = ";
237 switch (OpC) {
238 case SimpleRemoteEPCOpcode::Hangup:
239 dbgs() << "Hangup";
240 assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
241 assert(!TagAddr && "Non-zero TagAddr for Hangup?");
242 break;
243 case SimpleRemoteEPCOpcode::Result:
244 dbgs() << "Result";
245 assert(!TagAddr && "Non-zero TagAddr for Result?");
246 break;
247 case SimpleRemoteEPCOpcode::CallWrapper:
248 dbgs() << "CallWrapper";
249 break;
250 default:
251 llvm_unreachable("Invalid opcode");
252 }
253 dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
254 << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
255 << " bytes\n";
256 });
257 auto Err = T->sendMessage(OpC, SeqNo, TagAddr, ArgBytes);
258 LLVM_DEBUG({
259 if (Err)
260 dbgs() << " \\--> SimpleRemoteEPC::sendMessage failed\n";
261 });
262 return Err;
263}
264
265Error SimpleRemoteEPC::handleSetup(uint64_t SeqNo, ExecutorAddr TagAddr,
266 shared::WrapperFunctionBuffer ArgBytes) {
267 if (SeqNo != 0)
268 return make_error<StringError>(Args: "Setup packet SeqNo not zero",
269 Args: inconvertibleErrorCode());
270
271 if (TagAddr)
272 return make_error<StringError>(Args: "Setup packet TagAddr not zero",
273 Args: inconvertibleErrorCode());
274
275 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
276 auto I = PendingCallWrapperResults.find(Val: 0);
277 assert(PendingCallWrapperResults.size() == 1 &&
278 I != PendingCallWrapperResults.end() &&
279 "Setup message handler not connectly set up");
280 auto SetupMsgHandler = std::move(I->second);
281 PendingCallWrapperResults.erase(I);
282
283 auto WFR =
284 shared::WrapperFunctionBuffer::copyFrom(Source: ArgBytes.data(), Size: ArgBytes.size());
285 SetupMsgHandler(std::move(WFR));
286 return Error::success();
287}
288
289Error SimpleRemoteEPC::setup() {
290 using namespace SimpleRemoteEPCDefaultBootstrapSymbolNames;
291
292 std::promise<MSVCPExpected<SimpleRemoteEPCExecutorInfo>> EIP;
293 auto EIF = EIP.get_future();
294
295 // Prepare a handler for the setup packet.
296 PendingCallWrapperResults[0] =
297 RunInPlace()(
298 [&](shared::WrapperFunctionBuffer SetupMsgBytes) {
299 if (const char *ErrMsg = SetupMsgBytes.getOutOfBandError()) {
300 EIP.set_value(
301 make_error<StringError>(Args&: ErrMsg, Args: inconvertibleErrorCode()));
302 return;
303 }
304 using SPSSerialize =
305 shared::SPSArgList<shared::SPSSimpleRemoteEPCExecutorInfo>;
306 shared::SPSInputBuffer IB(SetupMsgBytes.data(), SetupMsgBytes.size());
307 SimpleRemoteEPCExecutorInfo EI;
308 if (SPSSerialize::deserialize(IB, Arg&: EI))
309 EIP.set_value(EI);
310 else
311 EIP.set_value(make_error<StringError>(
312 Args: "Could not deserialize setup message", Args: inconvertibleErrorCode()));
313 });
314
315 // Start the transport.
316 if (auto Err = T->start())
317 return Err;
318
319 // Wait for setup packet to arrive.
320 auto EI = EIF.get();
321 if (!EI) {
322 T->disconnect();
323 return EI.takeError();
324 }
325
326 LLVM_DEBUG({
327 dbgs() << "SimpleRemoteEPC received setup message:\n"
328 << " Triple: " << EI->TargetTriple << "\n"
329 << " Page size: " << EI->PageSize << "\n"
330 << " Bootstrap map" << (EI->BootstrapMap.empty() ? " empty" : ":")
331 << "\n";
332 for (const auto &KV : EI->BootstrapMap)
333 dbgs() << " " << KV.first() << ": " << KV.second.size()
334 << "-byte SPS encoded buffer\n";
335 dbgs() << " Bootstrap symbols"
336 << (EI->BootstrapSymbols.empty() ? " empty" : ":") << "\n";
337 for (const auto &KV : EI->BootstrapSymbols)
338 dbgs() << " " << KV.first() << ": " << KV.second << "\n";
339 });
340 TargetTriple = Triple(EI->TargetTriple);
341 PageSize = EI->PageSize;
342 BootstrapMap = std::move(EI->BootstrapMap);
343 BootstrapSymbols = std::move(EI->BootstrapSymbols);
344
345 BootstrapSymbols[rt::DispatchName] = BootstrapSymbols[DispatchFnName];
346 BootstrapSymbols[rt::DispatchCtxName] =
347 BootstrapSymbols[ExecutorSessionObjectName];
348
349 return Error::success();
350}
351
352Error SimpleRemoteEPC::handleResult(uint64_t SeqNo, ExecutorAddr TagAddr,
353 shared::WrapperFunctionBuffer ArgBytes) {
354 IncomingWFRHandler SendResult;
355
356 if (TagAddr)
357 return make_error<StringError>(Args: "Unexpected TagAddr in result message",
358 Args: inconvertibleErrorCode());
359
360 {
361 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
362 auto I = PendingCallWrapperResults.find(Val: SeqNo);
363 if (I == PendingCallWrapperResults.end())
364 return make_error<StringError>(Args: "No call for sequence number " +
365 Twine(SeqNo),
366 Args: inconvertibleErrorCode());
367 SendResult = std::move(I->second);
368 PendingCallWrapperResults.erase(I);
369 releaseSeqNo(SeqNo);
370 }
371
372 auto WFR =
373 shared::WrapperFunctionBuffer::copyFrom(Source: ArgBytes.data(), Size: ArgBytes.size());
374 SendResult(std::move(WFR));
375 return Error::success();
376}
377
378void SimpleRemoteEPC::handleCallWrapper(
379 uint64_t RemoteSeqNo, ExecutorAddr TagAddr,
380 shared::WrapperFunctionBuffer ArgBytes) {
381 assert(ES && "No ExecutionSession attached");
382 D->dispatch(T: makeGenericNamedTask(
383 Fn: [this, RemoteSeqNo, TagAddr, ArgBytes = std::move(ArgBytes)]() mutable {
384 ES->runJITDispatchHandler(
385 SendResult: [this, RemoteSeqNo](shared::WrapperFunctionBuffer WFR) {
386 if (auto Err =
387 sendMessage(OpC: SimpleRemoteEPCOpcode::Result, SeqNo: RemoteSeqNo,
388 TagAddr: ExecutorAddr(), ArgBytes: {WFR.data(), WFR.size()}))
389 getExecutionSession().reportError(Err: std::move(Err));
390 },
391 HandlerFnTagAddr: TagAddr, ArgBytes: std::move(ArgBytes));
392 },
393 Desc: "callWrapper task"));
394}
395
396Error SimpleRemoteEPC::handleHangup(shared::WrapperFunctionBuffer ArgBytes) {
397 return decodeHangupPayload(Payload: std::move(ArgBytes));
398}
399
400} // end namespace orc
401} // end namespace llvm
402