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/Mangler.h"
17#include "llvm/ExecutionEngine/Orc/Shared/OrcRTBridge.h"
18#include "llvm/Support/FormatVariadic.h"
19
20#define DEBUG_TYPE "orc"
21
22namespace llvm {
23namespace orc {
24
25SimpleRemoteEPC::~SimpleRemoteEPC() {
26#ifndef NDEBUG
27 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
28 assert(Disconnected && "Destroyed without disconnection");
29#endif // NDEBUG
30}
31
32Expected<int32_t> SimpleRemoteEPC::runAsMain(ExecutorAddr MainFnAddr,
33 ArrayRef<std::string> Args) {
34 if (!CallMain)
35 if (auto Err =
36 lookupAndApply(JD&: getExecutionSession().getBootstrapJITDylib(),
37 PrepareFns: recordProxy<sps::CallMainProxySpec>(P: &CallMain)))
38 return Err;
39
40 auto Result = CallMain(getExecutionSession(), MainFnAddr, Args);
41 if (!Result)
42 return Result.takeError();
43 return *Result;
44}
45
46void SimpleRemoteEPC::callWrapperAsync(ExecutorAddr WrapperFnAddr,
47 IncomingWFRHandler OnComplete,
48 ArrayRef<char> ArgBuffer) {
49 uint64_t SeqNo;
50 {
51 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
52 SeqNo = getNextSeqNo();
53 assert(!PendingCallWrapperResults.count(SeqNo) && "SeqNo already in use");
54 PendingCallWrapperResults[SeqNo] = std::move(OnComplete);
55 }
56
57 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::CallWrapper, SeqNo,
58 TagAddr: WrapperFnAddr, ArgBytes: ArgBuffer)) {
59 IncomingWFRHandler H;
60
61 // We just registered OnComplete, but there may be a race between this
62 // thread returning from sendMessage and handleDisconnect being called from
63 // the transport's listener thread. If handleDisconnect gets there first
64 // then it will have failed 'H' for us. If we get there first (or if
65 // handleDisconnect already ran) then we need to take care of it.
66 {
67 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
68 auto I = PendingCallWrapperResults.find(Val: SeqNo);
69 if (I != PendingCallWrapperResults.end()) {
70 H = std::move(I->second);
71 PendingCallWrapperResults.erase(I);
72 }
73 }
74
75 if (H)
76 H(shared::WrapperFunctionBuffer::createOutOfBandError(Msg: "disconnecting"));
77
78 getExecutionSession().reportError(Err: std::move(Err));
79 }
80}
81
82Expected<std::unique_ptr<jitlink::JITLinkMemoryManager>>
83SimpleRemoteEPC::createDefaultMemoryManager() {
84 return sps::createEPCGenericJITLinkMemoryManager(ES&: getExecutionSession());
85}
86
87Expected<std::unique_ptr<DylibManager>>
88SimpleRemoteEPC::createDefaultDylibMgr() {
89 return sps::createEPCGenericDylibManager(ES&: getExecutionSession());
90}
91
92Expected<std::unique_ptr<MemoryAccess>>
93SimpleRemoteEPC::createDefaultMemoryAccess() {
94 return sps::createEPCGenericMemoryAccess(ES&: getExecutionSession());
95}
96
97Error SimpleRemoteEPC::disconnect() {
98 // disconnect is idempotent, so the first caller owns the hangup. There is
99 // also nothing to announce to an executor that has already announced its own
100 // departure.
101 bool SendHangup = false;
102 {
103 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
104 SendHangup = !LocalHangup && !RemoteHangup;
105 LocalHangup = true;
106 }
107
108 // Tell the executor we're going away, so that it can distinguish this from
109 // losing us unexpectedly. Best-effort: if the send fails there is nothing to
110 // do but tear down anyway, and the executor will report the disconnection as
111 // unexpected. A locally requested disconnect is orderly, so the hangup
112 // carries a success value.
113 if (SendHangup) {
114 auto Payload = encodeHangupPayload(Err: Error::success());
115 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::Hangup, SeqNo: 0, TagAddr: ExecutorAddr(),
116 ArgBytes: {Payload.data(), Payload.size()}))
117 consumeError(Err: std::move(Err));
118 }
119
120 T->disconnect();
121 D->shutdown();
122 std::unique_lock<std::mutex> Lock(SimpleRemoteEPCMutex);
123 DisconnectCV.wait(lock&: Lock, p: [this] { return Disconnected; });
124 return std::move(DisconnectErr);
125}
126
127Expected<SimpleRemoteEPCTransportClient::HandleMessageAction>
128SimpleRemoteEPC::handleMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo,
129 ExecutorAddr TagAddr,
130 shared::WrapperFunctionBuffer ArgBytes) {
131
132 LLVM_DEBUG({
133 dbgs() << "SimpleRemoteEPC::handleMessage: opc = ";
134 switch (OpC) {
135 case SimpleRemoteEPCOpcode::Setup:
136 dbgs() << "Setup";
137 assert(SeqNo == 0 && "Non-zero SeqNo for Setup?");
138 assert(!TagAddr && "Non-zero TagAddr for Setup?");
139 break;
140 case SimpleRemoteEPCOpcode::Hangup:
141 dbgs() << "Hangup";
142 assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
143 assert(!TagAddr && "Non-zero TagAddr for Hangup?");
144 break;
145 case SimpleRemoteEPCOpcode::Result:
146 dbgs() << "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 break;
246 case SimpleRemoteEPCOpcode::CallWrapper:
247 dbgs() << "CallWrapper";
248 break;
249 default:
250 llvm_unreachable("Invalid opcode");
251 }
252 dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
253 << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
254 << " bytes\n";
255 });
256 auto Err = T->sendMessage(OpC, SeqNo, TagAddr, ArgBytes);
257 LLVM_DEBUG({
258 if (Err)
259 dbgs() << " \\--> SimpleRemoteEPC::sendMessage failed\n";
260 });
261 return Err;
262}
263
264Error SimpleRemoteEPC::handleSetup(uint64_t SeqNo, ExecutorAddr TagAddr,
265 shared::WrapperFunctionBuffer ArgBytes) {
266 if (SeqNo != 0)
267 return make_error<StringError>(Args: "Setup packet SeqNo not zero",
268 Args: inconvertibleErrorCode());
269
270 if (TagAddr)
271 return make_error<StringError>(Args: "Setup packet TagAddr not zero",
272 Args: inconvertibleErrorCode());
273
274 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
275 auto I = PendingCallWrapperResults.find(Val: 0);
276 assert(PendingCallWrapperResults.size() == 1 &&
277 I != PendingCallWrapperResults.end() &&
278 "Setup message handler not connectly set up");
279 auto SetupMsgHandler = std::move(I->second);
280 PendingCallWrapperResults.erase(I);
281
282 auto WFR =
283 shared::WrapperFunctionBuffer::copyFrom(Source: ArgBytes.data(), Size: ArgBytes.size());
284 SetupMsgHandler(std::move(WFR));
285 return Error::success();
286}
287
288Error SimpleRemoteEPC::setup() {
289 using namespace SimpleRemoteEPCDefaultBootstrapSymbolNames;
290
291 std::promise<MSVCPExpected<SimpleRemoteEPCExecutorInfo>> EIP;
292 auto EIF = EIP.get_future();
293
294 // Prepare a handler for the setup packet.
295 PendingCallWrapperResults[0] =
296 RunInPlace()(
297 [&](shared::WrapperFunctionBuffer SetupMsgBytes) {
298 if (const char *ErrMsg = SetupMsgBytes.getOutOfBandError()) {
299 EIP.set_value(
300 make_error<StringError>(Args&: ErrMsg, Args: inconvertibleErrorCode()));
301 return;
302 }
303 using SPSSerialize =
304 shared::SPSArgList<shared::SPSSimpleRemoteEPCExecutorInfo>;
305 shared::SPSInputBuffer IB(SetupMsgBytes.data(), SetupMsgBytes.size());
306 SimpleRemoteEPCExecutorInfo EI;
307 if (SPSSerialize::deserialize(IB, Arg&: EI))
308 EIP.set_value(EI);
309 else
310 EIP.set_value(make_error<StringError>(
311 Args: "Could not deserialize setup message", Args: inconvertibleErrorCode()));
312 });
313
314 // Start the transport.
315 if (auto Err = T->start())
316 return Err;
317
318 // Wait for setup packet to arrive.
319 auto EI = EIF.get();
320 if (!EI) {
321 T->disconnect();
322 return EI.takeError();
323 }
324
325 LLVM_DEBUG({
326 dbgs() << "SimpleRemoteEPC received setup message:\n"
327 << " Triple: " << EI->TargetTriple << "\n"
328 << " Page size: " << EI->PageSize << "\n"
329 << " Bootstrap map" << (EI->BootstrapMap.empty() ? " empty" : ":")
330 << "\n";
331 for (const auto &KV : EI->BootstrapMap)
332 dbgs() << " " << KV.first() << ": " << KV.second.size()
333 << "-byte SPS encoded buffer\n";
334 dbgs() << " Bootstrap symbols"
335 << (EI->BootstrapSymbols.empty() ? " empty" : ":") << "\n";
336 for (const auto &KV : EI->BootstrapSymbols)
337 dbgs() << " " << KV.first() << ": " << KV.second << "\n";
338 });
339 TargetTriple = Triple(EI->TargetTriple);
340 PageSize = EI->PageSize;
341 BootstrapMap = std::move(EI->BootstrapMap);
342 BootstrapSymbols = std::move(EI->BootstrapSymbols);
343
344 Mangler Mangle(getTargetTriple());
345 BootstrapSymbols[Mangle.mangledCopy(Name: rt::DispatchName)] =
346 BootstrapSymbols[DispatchFnName];
347 BootstrapSymbols[Mangle.mangledCopy(Name: rt::DispatchCtxName)] =
348 BootstrapSymbols[ExecutorSessionObjectName];
349
350 return Error::success();
351}
352
353Error SimpleRemoteEPC::handleResult(uint64_t SeqNo, ExecutorAddr TagAddr,
354 shared::WrapperFunctionBuffer ArgBytes) {
355 IncomingWFRHandler SendResult;
356
357 auto WFR = decodeResultMessage(TagAddr, Payload: std::move(ArgBytes));
358 if (!WFR)
359 return WFR.takeError();
360
361 {
362 std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
363 auto I = PendingCallWrapperResults.find(Val: SeqNo);
364 if (I == PendingCallWrapperResults.end())
365 return make_error<StringError>(Args: "No call for sequence number " +
366 Twine(SeqNo),
367 Args: inconvertibleErrorCode());
368 SendResult = std::move(I->second);
369 PendingCallWrapperResults.erase(I);
370 releaseSeqNo(SeqNo);
371 }
372
373 SendResult(std::move(*WFR));
374 return Error::success();
375}
376
377void SimpleRemoteEPC::handleCallWrapper(
378 uint64_t RemoteSeqNo, ExecutorAddr TagAddr,
379 shared::WrapperFunctionBuffer ArgBytes) {
380 assert(ES && "No ExecutionSession attached");
381 D->dispatch(T: makeGenericNamedTask(
382 Fn: [this, RemoteSeqNo, TagAddr, ArgBytes = std::move(ArgBytes)]() mutable {
383 ES->runJITDispatchHandler(
384 SendResult: [this, RemoteSeqNo](shared::WrapperFunctionBuffer WFR) {
385 auto [ResultTag, Payload] = encodeResultMessage(ResultBytes: std::move(WFR));
386 if (auto Err =
387 sendMessage(OpC: SimpleRemoteEPCOpcode::Result, SeqNo: RemoteSeqNo,
388 TagAddr: ResultTag, ArgBytes: {Payload.data(), Payload.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