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 return SimpleRemoteEPCTransportClient::EndSession;
105 case SimpleRemoteEPCOpcode::Result:
106 if (auto Err = handleResult(SeqNo, TagAddr, ArgBytes: std::move(ArgBytes)))
107 return std::move(Err);
108 break;
109 case SimpleRemoteEPCOpcode::CallWrapper:
110 handleCallWrapper(RemoteSeqNo: SeqNo, TagAddr, ArgBytes: std::move(ArgBytes));
111 break;
112 }
113 return ContinueSession;
114}
115
116Error SimpleRemoteEPCServer::waitForDisconnect() {
117 std::unique_lock<std::mutex> Lock(ServerStateMutex);
118 ShutdownCV.wait(lock&: Lock, p: [this]() { return RunState == ServerShutDown; });
119 return std::move(ShutdownErr);
120}
121
122void SimpleRemoteEPCServer::handleDisconnect(Error Err) {
123 PendingJITDispatchResultsMap TmpPending;
124
125 {
126 std::lock_guard<std::mutex> Lock(ServerStateMutex);
127 std::swap(a&: TmpPending, b&: PendingJITDispatchResults);
128 RunState = ServerShuttingDown;
129 }
130
131 // Send out-of-band errors to any waiting threads.
132 for (auto &KV : TmpPending)
133 KV.second->set_value(
134 shared::WrapperFunctionBuffer::createOutOfBandError(Msg: "disconnecting"));
135
136 // Wait for dispatcher to clear.
137 D->shutdown();
138
139 // Shut down services.
140 while (!Services.empty()) {
141 ShutdownErr =
142 joinErrors(E1: std::move(ShutdownErr), E2: Services.back()->shutdown());
143 Services.pop_back();
144 }
145
146 std::lock_guard<std::mutex> Lock(ServerStateMutex);
147 ShutdownErr = joinErrors(E1: std::move(ShutdownErr), E2: std::move(Err));
148 RunState = ServerShutDown;
149 ShutdownCV.notify_all();
150}
151
152Error SimpleRemoteEPCServer::sendMessage(SimpleRemoteEPCOpcode OpC,
153 uint64_t SeqNo, ExecutorAddr TagAddr,
154 ArrayRef<char> ArgBytes) {
155
156 LLVM_DEBUG({
157 dbgs() << "SimpleRemoteEPCServer::sendMessage: opc = ";
158 switch (OpC) {
159 case SimpleRemoteEPCOpcode::Setup:
160 dbgs() << "Setup";
161 assert(SeqNo == 0 && "Non-zero SeqNo for Setup?");
162 assert(!TagAddr && "Non-zero TagAddr for Setup?");
163 break;
164 case SimpleRemoteEPCOpcode::Hangup:
165 dbgs() << "Hangup";
166 assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
167 assert(!TagAddr && "Non-zero TagAddr for Hangup?");
168 break;
169 case SimpleRemoteEPCOpcode::Result:
170 dbgs() << "Result";
171 assert(!TagAddr && "Non-zero TagAddr for Result?");
172 break;
173 case SimpleRemoteEPCOpcode::CallWrapper:
174 dbgs() << "CallWrapper";
175 break;
176 }
177 dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
178 << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
179 << " bytes\n";
180 });
181 auto Err = T->sendMessage(OpC, SeqNo, TagAddr, ArgBytes);
182 LLVM_DEBUG({
183 if (Err)
184 dbgs() << " \\--> SimpleRemoteEPC::sendMessage failed\n";
185 });
186 return Err;
187}
188
189Error SimpleRemoteEPCServer::sendSetupMessage(
190 StringMap<std::vector<char>> BootstrapMap,
191 StringMap<ExecutorAddr> BootstrapSymbols) {
192
193 using namespace SimpleRemoteEPCDefaultBootstrapSymbolNames;
194
195 SimpleRemoteEPCExecutorInfo EI;
196 EI.TargetTriple = sys::getProcessTriple();
197 if (auto PageSize = sys::Process::getPageSize())
198 EI.PageSize = *PageSize;
199 else
200 return PageSize.takeError();
201 EI.BootstrapMap = std::move(BootstrapMap);
202 EI.BootstrapSymbols = std::move(BootstrapSymbols);
203
204 assert(!EI.BootstrapSymbols.count(ExecutorSessionObjectName) &&
205 "Dispatch context name should not be set");
206 assert(!EI.BootstrapSymbols.count(DispatchFnName) &&
207 "Dispatch function name should not be set");
208 EI.BootstrapSymbols[ExecutorSessionObjectName] = ExecutorAddr::fromPtr(Ptr: this);
209 EI.BootstrapSymbols[DispatchFnName] = ExecutorAddr::fromPtr(Ptr: jitDispatchEntry);
210 addDefaultBootstrapValuesForHostProcess(BootstrapMap&: EI.BootstrapMap, BootstrapSymbols&: EI.BootstrapSymbols);
211
212 using SPSSerialize =
213 shared::SPSArgList<shared::SPSSimpleRemoteEPCExecutorInfo>;
214 auto SetupPacketBytes =
215 shared::WrapperFunctionBuffer::allocate(Size: SPSSerialize::size(Arg: EI));
216 shared::SPSOutputBuffer OB(SetupPacketBytes.data(), SetupPacketBytes.size());
217 if (!SPSSerialize::serialize(OB, Arg: EI))
218 return make_error<StringError>(Args: "Could not send setup packet",
219 Args: inconvertibleErrorCode());
220
221 return sendMessage(OpC: SimpleRemoteEPCOpcode::Setup, SeqNo: 0, TagAddr: ExecutorAddr(),
222 ArgBytes: {SetupPacketBytes.data(), SetupPacketBytes.size()});
223}
224
225Error SimpleRemoteEPCServer::handleResult(
226 uint64_t SeqNo, ExecutorAddr TagAddr,
227 shared::WrapperFunctionBuffer ArgBytes) {
228 std::promise<shared::WrapperFunctionBuffer> *P = nullptr;
229 {
230 std::lock_guard<std::mutex> Lock(ServerStateMutex);
231 auto I = PendingJITDispatchResults.find(Val: SeqNo);
232 if (I == PendingJITDispatchResults.end())
233 return make_error<StringError>(Args: "No call for sequence number " +
234 Twine(SeqNo),
235 Args: inconvertibleErrorCode());
236 P = I->second;
237 PendingJITDispatchResults.erase(I);
238 releaseSeqNo(SeqNo);
239 }
240 auto R = shared::WrapperFunctionBuffer::allocate(Size: ArgBytes.size());
241 memcpy(dest: R.data(), src: ArgBytes.data(), n: ArgBytes.size());
242 P->set_value(std::move(R));
243 return Error::success();
244}
245
246void SimpleRemoteEPCServer::handleCallWrapper(
247 uint64_t RemoteSeqNo, ExecutorAddr TagAddr,
248 shared::WrapperFunctionBuffer ArgBytes) {
249 D->dispatch(Work: [this, RemoteSeqNo, TagAddr, ArgBytes = std::move(ArgBytes)]() {
250 using WrapperFnTy =
251 shared::CWrapperFunctionBuffer (*)(const char *, size_t);
252 auto *Fn = TagAddr.toPtr<WrapperFnTy>();
253 shared::WrapperFunctionBuffer ResultBytes(
254 Fn(ArgBytes.data(), ArgBytes.size()));
255 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::Result, SeqNo: RemoteSeqNo,
256 TagAddr: ExecutorAddr(),
257 ArgBytes: {ResultBytes.data(), ResultBytes.size()}))
258 ReportError(std::move(Err));
259 });
260}
261
262shared::WrapperFunctionBuffer
263SimpleRemoteEPCServer::doJITDispatch(const void *FnTag, const char *ArgData,
264 size_t ArgSize) {
265 uint64_t SeqNo;
266 std::promise<shared::WrapperFunctionBuffer> ResultP;
267 auto ResultF = ResultP.get_future();
268 {
269 std::lock_guard<std::mutex> Lock(ServerStateMutex);
270 if (RunState != ServerRunning)
271 return shared::WrapperFunctionBuffer::createOutOfBandError(
272 Msg: "jit_dispatch not available (EPC server shut down)");
273
274 SeqNo = getNextSeqNo();
275 assert(!PendingJITDispatchResults.count(SeqNo) && "SeqNo already in use");
276 PendingJITDispatchResults[SeqNo] = &ResultP;
277 }
278
279 if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::CallWrapper, SeqNo,
280 TagAddr: ExecutorAddr::fromPtr(Ptr: FnTag), ArgBytes: {ArgData, ArgSize}))
281 ReportError(std::move(Err));
282
283 return ResultF.get();
284}
285
286shared::CWrapperFunctionBuffer
287SimpleRemoteEPCServer::jitDispatchEntry(void *DispatchCtx, const void *FnTag,
288 const char *ArgData, size_t ArgSize) {
289 return reinterpret_cast<SimpleRemoteEPCServer *>(DispatchCtx)
290 ->doJITDispatch(FnTag, ArgData, ArgSize)
291 .release();
292}
293
294} // end namespace orc
295} // end namespace llvm
296