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/Shared/TargetProcessControlTypes.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 | |
22 | using namespace llvm::orc::shared; |
23 | |
24 | namespace llvm { |
25 | namespace orc { |
26 | |
27 | ExecutorBootstrapService::~ExecutorBootstrapService() = default; |
28 | |
29 | SimpleRemoteEPCServer::Dispatcher::~Dispatcher() = default; |
30 | |
31 | #if LLVM_ENABLE_THREADS |
32 | void 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 | |
49 | void 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 | |
56 | StringMap<ExecutorAddr> SimpleRemoteEPCServer::defaultBootstrapSymbols() { |
57 | StringMap<ExecutorAddr> DBS; |
58 | rt_bootstrap::addTo(M&: DBS); |
59 | return DBS; |
60 | } |
61 | |
62 | Expected<SimpleRemoteEPCTransportClient::HandleMessageAction> |
63 | SimpleRemoteEPCServer::handleMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo, |
64 | ExecutorAddr TagAddr, |
65 | SimpleRemoteEPCArgBytesVector 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 | |
116 | Error 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 | |
122 | void 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::WrapperFunctionResult::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 | |
152 | Error 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 | |
189 | Error SimpleRemoteEPCServer::sendSetupMessage( |
190 | StringMap<std::vector<char>> BootstrapMap, |
191 | StringMap<ExecutorAddr> BootstrapSymbols) { |
192 | |
193 | using namespace SimpleRemoteEPCDefaultBootstrapSymbolNames; |
194 | |
195 | std::vector<char> SetupPacket; |
196 | SimpleRemoteEPCExecutorInfo EI; |
197 | EI.TargetTriple = sys::getProcessTriple(); |
198 | if (auto PageSize = sys::Process::getPageSize()) |
199 | EI.PageSize = *PageSize; |
200 | else |
201 | return PageSize.takeError(); |
202 | EI.BootstrapMap = std::move(BootstrapMap); |
203 | EI.BootstrapSymbols = std::move(BootstrapSymbols); |
204 | |
205 | assert(!EI.BootstrapSymbols.count(ExecutorSessionObjectName) && |
206 | "Dispatch context name should not be set" ); |
207 | assert(!EI.BootstrapSymbols.count(DispatchFnName) && |
208 | "Dispatch function name should not be set" ); |
209 | EI.BootstrapSymbols[ExecutorSessionObjectName] = ExecutorAddr::fromPtr(Ptr: this); |
210 | EI.BootstrapSymbols[DispatchFnName] = ExecutorAddr::fromPtr(Ptr: jitDispatchEntry); |
211 | EI.BootstrapSymbols[rt::RegisterEHFrameSectionWrapperName] = |
212 | ExecutorAddr::fromPtr(Ptr: &llvm_orc_registerEHFrameSectionWrapper); |
213 | EI.BootstrapSymbols[rt::DeregisterEHFrameSectionWrapperName] = |
214 | ExecutorAddr::fromPtr(Ptr: &llvm_orc_deregisterEHFrameSectionWrapper); |
215 | |
216 | using SPSSerialize = |
217 | shared::SPSArgList<shared::SPSSimpleRemoteEPCExecutorInfo>; |
218 | auto SetupPacketBytes = |
219 | shared::WrapperFunctionResult::allocate(Size: SPSSerialize::size(Arg: EI)); |
220 | shared::SPSOutputBuffer OB(SetupPacketBytes.data(), SetupPacketBytes.size()); |
221 | if (!SPSSerialize::serialize(OB, Arg: EI)) |
222 | return make_error<StringError>(Args: "Could not send setup packet" , |
223 | Args: inconvertibleErrorCode()); |
224 | |
225 | return sendMessage(OpC: SimpleRemoteEPCOpcode::Setup, SeqNo: 0, TagAddr: ExecutorAddr(), |
226 | ArgBytes: {SetupPacketBytes.data(), SetupPacketBytes.size()}); |
227 | } |
228 | |
229 | Error SimpleRemoteEPCServer::handleResult( |
230 | uint64_t SeqNo, ExecutorAddr TagAddr, |
231 | SimpleRemoteEPCArgBytesVector ArgBytes) { |
232 | std::promise<shared::WrapperFunctionResult> *P = nullptr; |
233 | { |
234 | std::lock_guard<std::mutex> Lock(ServerStateMutex); |
235 | auto I = PendingJITDispatchResults.find(Val: SeqNo); |
236 | if (I == PendingJITDispatchResults.end()) |
237 | return make_error<StringError>(Args: "No call for sequence number " + |
238 | Twine(SeqNo), |
239 | Args: inconvertibleErrorCode()); |
240 | P = I->second; |
241 | PendingJITDispatchResults.erase(I); |
242 | releaseSeqNo(SeqNo); |
243 | } |
244 | auto R = shared::WrapperFunctionResult::allocate(Size: ArgBytes.size()); |
245 | memcpy(dest: R.data(), src: ArgBytes.data(), n: ArgBytes.size()); |
246 | P->set_value(std::move(R)); |
247 | return Error::success(); |
248 | } |
249 | |
250 | void SimpleRemoteEPCServer::handleCallWrapper( |
251 | uint64_t RemoteSeqNo, ExecutorAddr TagAddr, |
252 | SimpleRemoteEPCArgBytesVector ArgBytes) { |
253 | D->dispatch(Work: [this, RemoteSeqNo, TagAddr, ArgBytes = std::move(ArgBytes)]() { |
254 | using WrapperFnTy = |
255 | shared::CWrapperFunctionResult (*)(const char *, size_t); |
256 | auto *Fn = TagAddr.toPtr<WrapperFnTy>(); |
257 | shared::WrapperFunctionResult ResultBytes( |
258 | Fn(ArgBytes.data(), ArgBytes.size())); |
259 | if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::Result, SeqNo: RemoteSeqNo, |
260 | TagAddr: ExecutorAddr(), |
261 | ArgBytes: {ResultBytes.data(), ResultBytes.size()})) |
262 | ReportError(std::move(Err)); |
263 | }); |
264 | } |
265 | |
266 | shared::WrapperFunctionResult |
267 | SimpleRemoteEPCServer::doJITDispatch(const void *FnTag, const char *ArgData, |
268 | size_t ArgSize) { |
269 | uint64_t SeqNo; |
270 | std::promise<shared::WrapperFunctionResult> ResultP; |
271 | auto ResultF = ResultP.get_future(); |
272 | { |
273 | std::lock_guard<std::mutex> Lock(ServerStateMutex); |
274 | if (RunState != ServerRunning) |
275 | return shared::WrapperFunctionResult::createOutOfBandError( |
276 | Msg: "jit_dispatch not available (EPC server shut down)" ); |
277 | |
278 | SeqNo = getNextSeqNo(); |
279 | assert(!PendingJITDispatchResults.count(SeqNo) && "SeqNo already in use" ); |
280 | PendingJITDispatchResults[SeqNo] = &ResultP; |
281 | } |
282 | |
283 | if (auto Err = sendMessage(OpC: SimpleRemoteEPCOpcode::CallWrapper, SeqNo, |
284 | TagAddr: ExecutorAddr::fromPtr(Ptr: FnTag), ArgBytes: {ArgData, ArgSize})) |
285 | ReportError(std::move(Err)); |
286 | |
287 | return ResultF.get(); |
288 | } |
289 | |
290 | shared::CWrapperFunctionResult |
291 | SimpleRemoteEPCServer::jitDispatchEntry(void *DispatchCtx, const void *FnTag, |
292 | const char *ArgData, size_t ArgSize) { |
293 | return reinterpret_cast<SimpleRemoteEPCServer *>(DispatchCtx) |
294 | ->doJITDispatch(FnTag, ArgData, ArgSize) |
295 | .release(); |
296 | } |
297 | |
298 | } // end namespace orc |
299 | } // end namespace llvm |
300 | |