1//===------- SimpleEPCServer.cpp - EPC over simple abstract channel -------===//
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/TargetProcess/SimpleRemoteEPCServer.h"
10
11#include "llvm/ExecutionEngine/Orc/Shared/OrcRTBridge.h"
12#include "llvm/ExecutionEngine/Orc/TargetProcess/DefaultHostBootstrapValues.h"
13#include "llvm/ExecutionEngine/Orc/TargetProcess/RegisterEHFrames.h"
14#include "llvm/Support/FormatVariadic.h"
15#include "llvm/Support/Process.h"
16#include "llvm/TargetParser/Host.h"
17
18#include "OrcRTBootstrap.h"
19
20#define DEBUG_TYPE "orc"
21
22using namespace llvm::orc::shared;
23
24namespace llvm {
25namespace orc {
26
27ExecutorBootstrapService::~ExecutorBootstrapService() = default;
28
29SimpleRemoteEPCServer::Dispatcher::~Dispatcher() = default;
30
31#if LLVM_ENABLE_THREADS
32void SimpleRemoteEPCServer::ThreadDispatcher::dispatch(
33 unique_function<void()> Work) {
34 {
35 std::lock_guard<std::mutex> Lock(DispatchMutex);
36 if (!Running)
37 return;
38 ++Outstanding;
39 }
40
41 std::thread([this, Work = std::move(Work)]() mutable {
42 Work();
43 std::lock_guard<std::mutex> Lock(DispatchMutex);
44 --Outstanding;
45 OutstandingCV.notify_all();
46 }).detach();
47}
48
49void SimpleRemoteEPCServer::ThreadDispatcher::shutdown() {
50 std::unique_lock<std::mutex> Lock(DispatchMutex);
51 Running = false;
52 OutstandingCV.wait(lock&: Lock, p: [this]() { return Outstanding == 0; });
53}
54#endif
55
56StringMap<ExecutorAddr> SimpleRemoteEPCServer::defaultBootstrapSymbols() {
57 StringMap<ExecutorAddr> DBS;
58 rt_bootstrap::addTo(M&: DBS);
59 return DBS;
60}
61
62Expected<SimpleRemoteEPCTransportClient::HandleMessageAction>
63SimpleRemoteEPCServer::handleMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo,
64 ExecutorAddr TagAddr,
65 shared::WrapperFunctionBuffer ArgBytes) {
66
67 LLVM_DEBUG({
68 dbgs() << "SimpleRemoteEPCServer::handleMessage: opc = ";
69 switch (OpC) {
70 case SimpleRemoteEPCOpcode::Setup:
71 dbgs() << "Setup";
72 assert(SeqNo == 0 && "Non-zero SeqNo for Setup?");
73 assert(!TagAddr && "Non-zero TagAddr for Setup?");
74 break;
75 case SimpleRemoteEPCOpcode::Hangup:
76 dbgs() << "Hangup";
77 assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
78 assert(!TagAddr && "Non-zero TagAddr for Hangup?");
79 break;
80 case SimpleRemoteEPCOpcode::Result:
81 dbgs() << "Result";
82 assert(!TagAddr && "Non-zero TagAddr for Result?");
83 break;
84 case SimpleRemoteEPCOpcode::CallWrapper:
85 dbgs() << "CallWrapper";
86 break;
87 }
88 dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
89 << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
90 << " bytes\n";
91 });
92
93 using UT = std::underlying_type_t<SimpleRemoteEPCOpcode>;
94 if (static_cast<UT>(OpC) > static_cast<UT>(SimpleRemoteEPCOpcode::LastOpC))
95 return make_error<StringError>(Args: "Unexpected opcode",
96 Args: inconvertibleErrorCode());
97
98 // TODO: Clean detach message?
99 switch (OpC) {
100 case SimpleRemoteEPCOpcode::Setup:
101 return make_error<StringError>(Args: "Unexpected Setup opcode",
102 Args: inconvertibleErrorCode());
103 case SimpleRemoteEPCOpcode::Hangup: {
104 {
105 std::lock_guard<std::mutex> Lock(ServerStateMutex);
106 RemoteHangup = true;
107 }
108 if (auto Err = decodeHangupPayload(Payload: std::move(ArgBytes)))
109 return std::move(Err);
110 return SimpleRemoteEPCTransportClient::EndSession;
111 }
112 case SimpleRemoteEPCOpcode::Result:
113 if (auto Err = handleResult(SeqNo, TagAddr, ArgBytes: std::move(ArgBytes)))
114 return std::move(Err);
115 break;
116 case SimpleRemoteEPCOpcode::CallWrapper:
117 handleCallWrapper(RemoteSeqNo: SeqNo, TagAddr, ArgBytes: std::move(ArgBytes));
118 break;
119 }
120 return ContinueSession;
121}
122
123Error SimpleRemoteEPCServer::waitForDisconnect() {
124 std::unique_lock<std::mutex> Lock(ServerStateMutex);
125 ShutdownCV.wait(lock&: Lock, p: [this]() { return RunState == ServerShutDown; });
126 return std::move(ShutdownErr);
127}
128
129void SimpleRemoteEPCServer::handleDisconnect(Error Err) {
130 PendingJITDispatchResultsMap TmpPending;
131
132 {
133 std::lock_guard<std::mutex> Lock(ServerStateMutex);
134 std::swap(a&: TmpPending, b&: PendingJITDispatchResults);
135 RunState = ServerShuttingDown;
136 }
137
138 // Send out-of-band errors to any waiting threads.
139 for (auto &KV : TmpPending)
140 KV.second->set_value(
141 shared::WrapperFunctionBuffer::createOutOfBandError(Msg: "disconnecting"));
142
143 // Wait for dispatcher to clear.
144 D->shutdown();
145
146 // Shut down services.
147 while (!Services.empty()) {
148 ShutdownErr =
149 joinErrors(E1: std::move(ShutdownErr), E2: Services.back()->shutdown());
150 Services.pop_back();
151 }
152
153 std::lock_guard<std::mutex> Lock(ServerStateMutex);
154
155 // The server never initiates a disconnection, so if the transport reported no
156 // error and no hangup arrived then the controller went away without telling
157 // us. The cause is not knowable from here -- it may have crashed, been
158 // killed, or become unreachable -- so report what was observed rather than a
159 // cause.
160 //
161 // A missing hangup is evidence, not proof: a hangup can also be lost in
162 // transit, since closing a TCP socket with unread data queued sends an RST,
163 // which can discard bytes the peer had already delivered. We accept that
164 // rather than draining the read side before closing -- the cost is a
165 // misleading diagnostic on a session that is ending regardless, whereas a
166 // drain risks stalling teardown on a peer that never closes.
167 Error DisconnectReason =
168 (!Err && !RemoteHangup)
169 ? make_error<StringError>(Args: "Connection closed without hangup",
170 Args: inconvertibleErrorCode())
171 : std::move(Err);
172
173 ShutdownErr = joinErrors(E1: std::move(ShutdownErr), E2: std::move(DisconnectReason));
174 RunState = ServerShutDown;
175 ShutdownCV.notify_all();
176}
177
178Error SimpleRemoteEPCServer::sendMessage(SimpleRemoteEPCOpcode OpC,
179 uint64_t SeqNo, ExecutorAddr TagAddr,
180 ArrayRef<char> ArgBytes) {
181
182 LLVM_DEBUG({
183 dbgs() << "SimpleRemoteEPCServer::sendMessage: opc = ";
184 switch (OpC) {
185 case SimpleRemoteEPCOpcode::Setup:
186 dbgs() << "Setup";
187 assert(SeqNo == 0 && "Non-zero SeqNo for Setup?");
188 assert(!TagAddr && "Non-zero TagAddr for Setup?");
189 break;
190 case SimpleRemoteEPCOpcode::Hangup:
191 dbgs() << "Hangup";
192 assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
193 assert(!TagAddr && "Non-zero TagAddr for Hangup?");
194 break;
195 case SimpleRemoteEPCOpcode::Result:
196 dbgs() << "Result";
197 assert(!TagAddr && "Non-zero TagAddr for Result?");
198 break;
199 case SimpleRemoteEPCOpcode::CallWrapper:
200 dbgs() << "CallWrapper";
201 break;
202 }
203 dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
204 << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
205 << " bytes\n";
206 });
207 auto Err = T->sendMessage(OpC, SeqNo, TagAddr, ArgBytes);
208 LLVM_DEBUG({
209 if (Err)
210 dbgs() << " \\--> SimpleRemoteEPC::sendMessage failed\n";
211 });
212 return Err;
213}
214
215Error SimpleRemoteEPCServer::sendSetupMessage(
216 StringMap<std::vector<char>> BootstrapMap,
217 StringMap<ExecutorAddr> BootstrapSymbols) {
218
219 using namespace SimpleRemoteEPCDefaultBootstrapSymbolNames;
220
221 SimpleRemoteEPCExecutorInfo EI;
222 EI.TargetTriple = sys::getProcessTriple();
223 if (auto PageSize = sys::Process::getPageSize())
224 EI.PageSize = *PageSize;
225 else
226 return PageSize.takeError();
227 EI.BootstrapMap = std::move(BootstrapMap);
228 EI.BootstrapSymbols = std::move(BootstrapSymbols);
229
230 assert(!EI.BootstrapSymbols.count(ExecutorSessionObjectName) &&
231 "Dispatch context name should not be set");
232 assert(!EI.BootstrapSymbols.count(DispatchFnName) &&
233 "Dispatch function name should not be set");
234 EI.BootstrapSymbols[ExecutorSessionObjectName] = ExecutorAddr::fromPtr(Ptr: this);
235 EI.BootstrapSymbols[DispatchFnName] = ExecutorAddr::fromPtr(Ptr: jitDispatchEntry);
236 addDefaultBootstrapValuesForHostProcess(BootstrapMap&: EI.BootstrapMap, BootstrapSymbols&: EI.BootstrapSymbols);
237
238 using SPSSerialize =
239 shared::SPSArgList<shared::SPSSimpleRemoteEPCExecutorInfo>;
240 auto SetupPacketBytes =
241 shared::WrapperFunctionBuffer::allocate(Size: SPSSerialize::size(Arg: EI));
242 shared::SPSOutputBuffer OB(SetupPacketBytes.data(), SetupPacketBytes.size());
243 if (!SPSSerialize::serialize(OB, Arg: EI))
244 return make_error<StringError>(Args: "Could not send setup packet",
245 Args: inconvertibleErrorCode());
246
247 return sendMessage(OpC: SimpleRemoteEPCOpcode::Setup, SeqNo: 0, TagAddr: ExecutorAddr(),
248 ArgBytes: {SetupPacketBytes.data(), SetupPacketBytes.size()});
249}
250
251Error SimpleRemoteEPCServer::handleResult(
252 uint64_t SeqNo, ExecutorAddr TagAddr,
253 shared::WrapperFunctionBuffer ArgBytes) {
254 std::promise<shared::WrapperFunctionBuffer> *P = nullptr;
255 {
256 std::lock_guard<std::mutex> Lock(ServerStateMutex);
257 auto I = PendingJITDispatchResults.find(Val: SeqNo);
258 if (I == PendingJITDispatchResults.end())
259 return make_error<StringError>(Args: "No call for sequence number " +
260 Twine(SeqNo),
261 Args: inconvertibleErrorCode());
262 P = I->second;
263 PendingJITDispatchResults.erase(I);
264 releaseSeqNo(SeqNo);
265 }
266 auto R = shared::WrapperFunctionBuffer::allocate(Size: ArgBytes.size());
267 memcpy(dest: R.data(), src: ArgBytes.data(), n: ArgBytes.size());
268 P->set_value(std::move(R));
269 return Error::success();
270}
271
272void SimpleRemoteEPCServer::handleCallWrapper(
273 uint64_t RemoteSeqNo, ExecutorAddr TagAddr,
274 shared::WrapperFunctionBuffer ArgBytes) {
275 D->dispatch(Work: [this, RemoteSeqNo, TagAddr, ArgBytes = std::move(ArgBytes)]() {
276 using WrapperFnTy =
277 shared::CWrapperFunctionBuffer (*)(const char *, size_t);
278 auto *Fn = TagAddr.toPtr<WrapperFnTy>();
279 shared::WrapperFunctionBuffer ResultBytes(
280 Fn(ArgBytes.data(), ArgBytes.size()));
281 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::Result, SeqNo: RemoteSeqNo,
282 TagAddr: ExecutorAddr(),
283 ArgBytes: {ResultBytes.data(), ResultBytes.size()}))
284 ReportError(std::move(Err));
285 });
286}
287
288shared::WrapperFunctionBuffer
289SimpleRemoteEPCServer::doJITDispatch(const void *FnTag, const char *ArgData,
290 size_t ArgSize) {
291 uint64_t SeqNo;
292 std::promise<shared::WrapperFunctionBuffer> ResultP;
293 auto ResultF = ResultP.get_future();
294 {
295 std::lock_guard<std::mutex> Lock(ServerStateMutex);
296 if (RunState != ServerRunning)
297 return shared::WrapperFunctionBuffer::createOutOfBandError(
298 Msg: "jit_dispatch not available (EPC server shut down)");
299
300 SeqNo = getNextSeqNo();
301 assert(!PendingJITDispatchResults.count(SeqNo) && "SeqNo already in use");
302 PendingJITDispatchResults[SeqNo] = &ResultP;
303 }
304
305 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::CallWrapper, SeqNo,
306 TagAddr: ExecutorAddr::fromPtr(Ptr: FnTag), ArgBytes: {ArgData, ArgSize}))
307 ReportError(std::move(Err));
308
309 return ResultF.get();
310}
311
312shared::CWrapperFunctionBuffer
313SimpleRemoteEPCServer::jitDispatchEntry(void *DispatchCtx, const void *FnTag,
314 const char *ArgData, size_t ArgSize) {
315 return reinterpret_cast<SimpleRemoteEPCServer *>(DispatchCtx)
316 ->doJITDispatch(FnTag, ArgData, ArgSize)
317 .release();
318}
319
320} // end namespace orc
321} // end namespace llvm
322