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