xref: /llvm-project/llvm/lib/ExecutionEngine/Orc/SimpleRemoteEPC.cpp (revision 1f4d91ecb8529678a3d3919d7523743bd21942ca)
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/EPCGenericJITLinkMemoryManager.h"
11 #include "llvm/ExecutionEngine/Orc/Shared/OrcRTBridge.h"
12 #include "llvm/Support/FormatVariadic.h"
13 
14 #define DEBUG_TYPE "orc"
15 
16 namespace llvm {
17 namespace orc {
18 
19 SimpleRemoteEPC::~SimpleRemoteEPC() {
20 #ifndef NDEBUG
21   std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
22   assert(Disconnected && "Destroyed without disconnection");
23 #endif // NDEBUG
24 }
25 
26 Expected<tpctypes::DylibHandle>
27 SimpleRemoteEPC::loadDylib(const char *DylibPath) {
28   return EPCDylibMgr->open(DylibPath, 0);
29 }
30 
31 /// Async helper to chain together calls to DylibMgr::lookupAsync to fulfill all
32 /// all the requests.
33 /// FIXME: The dylib manager should support multiple LookupRequests natively.
34 static void
35 lookupSymbolsAsyncHelper(EPCGenericDylibManager &DylibMgr,
36                          ArrayRef<DylibManager::LookupRequest> Request,
37                          std::vector<tpctypes::LookupResult> Result,
38                          DylibManager::SymbolLookupCompleteFn Complete) {
39   if (Request.empty())
40     return Complete(std::move(Result));
41 
42   auto &Element = Request.front();
43   DylibMgr.lookupAsync(Element.Handle, Element.Symbols,
44                        [&DylibMgr, Request, Complete = std::move(Complete),
45                         Result = std::move(Result)](auto R) mutable {
46                          if (!R)
47                            return Complete(R.takeError());
48                          Result.push_back({});
49                          Result.back().reserve(R->size());
50                          for (auto Addr : *R)
51                            Result.back().push_back(Addr);
52 
53                          lookupSymbolsAsyncHelper(
54                              DylibMgr, Request.drop_front(), std::move(Result),
55                              std::move(Complete));
56                        });
57 }
58 
59 void SimpleRemoteEPC::lookupSymbolsAsync(ArrayRef<LookupRequest> Request,
60                                          SymbolLookupCompleteFn Complete) {
61   lookupSymbolsAsyncHelper(*EPCDylibMgr, Request, {}, std::move(Complete));
62 }
63 
64 Expected<int32_t> SimpleRemoteEPC::runAsMain(ExecutorAddr MainFnAddr,
65                                              ArrayRef<std::string> Args) {
66   int64_t Result = 0;
67   if (auto Err = callSPSWrapper<rt::SPSRunAsMainSignature>(
68           RunAsMainAddr, Result, MainFnAddr, Args))
69     return std::move(Err);
70   return Result;
71 }
72 
73 Expected<int32_t> SimpleRemoteEPC::runAsVoidFunction(ExecutorAddr VoidFnAddr) {
74   int32_t Result = 0;
75   if (auto Err = callSPSWrapper<rt::SPSRunAsVoidFunctionSignature>(
76           RunAsVoidFunctionAddr, Result, VoidFnAddr))
77     return std::move(Err);
78   return Result;
79 }
80 
81 Expected<int32_t> SimpleRemoteEPC::runAsIntFunction(ExecutorAddr IntFnAddr,
82                                                     int Arg) {
83   int32_t Result = 0;
84   if (auto Err = callSPSWrapper<rt::SPSRunAsIntFunctionSignature>(
85           RunAsIntFunctionAddr, Result, IntFnAddr, Arg))
86     return std::move(Err);
87   return Result;
88 }
89 
90 void SimpleRemoteEPC::callWrapperAsync(ExecutorAddr WrapperFnAddr,
91                                        IncomingWFRHandler OnComplete,
92                                        ArrayRef<char> ArgBuffer) {
93   uint64_t SeqNo;
94   {
95     std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
96     SeqNo = getNextSeqNo();
97     assert(!PendingCallWrapperResults.count(SeqNo) && "SeqNo already in use");
98     PendingCallWrapperResults[SeqNo] = std::move(OnComplete);
99   }
100 
101   if (auto Err = sendMessage(SimpleRemoteEPCOpcode::CallWrapper, SeqNo,
102                              WrapperFnAddr, ArgBuffer)) {
103     IncomingWFRHandler H;
104 
105     // We just registered OnComplete, but there may be a race between this
106     // thread returning from sendMessage and handleDisconnect being called from
107     // the transport's listener thread. If handleDisconnect gets there first
108     // then it will have failed 'H' for us. If we get there first (or if
109     // handleDisconnect already ran) then we need to take care of it.
110     {
111       std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
112       auto I = PendingCallWrapperResults.find(SeqNo);
113       if (I != PendingCallWrapperResults.end()) {
114         H = std::move(I->second);
115         PendingCallWrapperResults.erase(I);
116       }
117     }
118 
119     if (H)
120       H(shared::WrapperFunctionResult::createOutOfBandError("disconnecting"));
121 
122     getExecutionSession().reportError(std::move(Err));
123   }
124 }
125 
126 Error SimpleRemoteEPC::disconnect() {
127   T->disconnect();
128   D->shutdown();
129   std::unique_lock<std::mutex> Lock(SimpleRemoteEPCMutex);
130   DisconnectCV.wait(Lock, [this] { return Disconnected; });
131   return std::move(DisconnectErr);
132 }
133 
134 Expected<SimpleRemoteEPCTransportClient::HandleMessageAction>
135 SimpleRemoteEPC::handleMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo,
136                                ExecutorAddr TagAddr,
137                                SimpleRemoteEPCArgBytesVector ArgBytes) {
138 
139   LLVM_DEBUG({
140     dbgs() << "SimpleRemoteEPC::handleMessage: opc = ";
141     switch (OpC) {
142     case SimpleRemoteEPCOpcode::Setup:
143       dbgs() << "Setup";
144       assert(SeqNo == 0 && "Non-zero SeqNo for Setup?");
145       assert(!TagAddr && "Non-zero TagAddr for Setup?");
146       break;
147     case SimpleRemoteEPCOpcode::Hangup:
148       dbgs() << "Hangup";
149       assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
150       assert(!TagAddr && "Non-zero TagAddr for Hangup?");
151       break;
152     case SimpleRemoteEPCOpcode::Result:
153       dbgs() << "Result";
154       assert(!TagAddr && "Non-zero TagAddr for Result?");
155       break;
156     case SimpleRemoteEPCOpcode::CallWrapper:
157       dbgs() << "CallWrapper";
158       break;
159     }
160     dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
161            << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
162            << " bytes\n";
163   });
164 
165   using UT = std::underlying_type_t<SimpleRemoteEPCOpcode>;
166   if (static_cast<UT>(OpC) > static_cast<UT>(SimpleRemoteEPCOpcode::LastOpC))
167     return make_error<StringError>("Unexpected opcode",
168                                    inconvertibleErrorCode());
169 
170   switch (OpC) {
171   case SimpleRemoteEPCOpcode::Setup:
172     if (auto Err = handleSetup(SeqNo, TagAddr, std::move(ArgBytes)))
173       return std::move(Err);
174     break;
175   case SimpleRemoteEPCOpcode::Hangup:
176     T->disconnect();
177     if (auto Err = handleHangup(std::move(ArgBytes)))
178       return std::move(Err);
179     return EndSession;
180   case SimpleRemoteEPCOpcode::Result:
181     if (auto Err = handleResult(SeqNo, TagAddr, std::move(ArgBytes)))
182       return std::move(Err);
183     break;
184   case SimpleRemoteEPCOpcode::CallWrapper:
185     handleCallWrapper(SeqNo, TagAddr, std::move(ArgBytes));
186     break;
187   }
188   return ContinueSession;
189 }
190 
191 void SimpleRemoteEPC::handleDisconnect(Error Err) {
192   LLVM_DEBUG({
193     dbgs() << "SimpleRemoteEPC::handleDisconnect: "
194            << (Err ? "failure" : "success") << "\n";
195   });
196 
197   PendingCallWrapperResultsMap TmpPending;
198 
199   {
200     std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
201     std::swap(TmpPending, PendingCallWrapperResults);
202   }
203 
204   for (auto &KV : TmpPending)
205     KV.second(
206         shared::WrapperFunctionResult::createOutOfBandError("disconnecting"));
207 
208   std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
209   DisconnectErr = joinErrors(std::move(DisconnectErr), std::move(Err));
210   Disconnected = true;
211   DisconnectCV.notify_all();
212 }
213 
214 Expected<std::unique_ptr<jitlink::JITLinkMemoryManager>>
215 SimpleRemoteEPC::createDefaultMemoryManager(SimpleRemoteEPC &SREPC) {
216   EPCGenericJITLinkMemoryManager::SymbolAddrs SAs;
217   if (auto Err = SREPC.getBootstrapSymbols(
218           {{SAs.Allocator, rt::SimpleExecutorMemoryManagerInstanceName},
219            {SAs.Reserve, rt::SimpleExecutorMemoryManagerReserveWrapperName},
220            {SAs.Finalize, rt::SimpleExecutorMemoryManagerFinalizeWrapperName},
221            {SAs.Deallocate,
222             rt::SimpleExecutorMemoryManagerDeallocateWrapperName}}))
223     return std::move(Err);
224 
225   return std::make_unique<EPCGenericJITLinkMemoryManager>(SREPC, SAs);
226 }
227 
228 Expected<std::unique_ptr<ExecutorProcessControl::MemoryAccess>>
229 SimpleRemoteEPC::createDefaultMemoryAccess(SimpleRemoteEPC &SREPC) {
230   return nullptr;
231 }
232 
233 Error SimpleRemoteEPC::sendMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo,
234                                    ExecutorAddr TagAddr,
235                                    ArrayRef<char> ArgBytes) {
236   assert(OpC != SimpleRemoteEPCOpcode::Setup &&
237          "SimpleRemoteEPC sending Setup message? That's the wrong direction.");
238 
239   LLVM_DEBUG({
240     dbgs() << "SimpleRemoteEPC::sendMessage: opc = ";
241     switch (OpC) {
242     case SimpleRemoteEPCOpcode::Hangup:
243       dbgs() << "Hangup";
244       assert(SeqNo == 0 && "Non-zero SeqNo for Hangup?");
245       assert(!TagAddr && "Non-zero TagAddr for Hangup?");
246       break;
247     case SimpleRemoteEPCOpcode::Result:
248       dbgs() << "Result";
249       assert(!TagAddr && "Non-zero TagAddr for Result?");
250       break;
251     case SimpleRemoteEPCOpcode::CallWrapper:
252       dbgs() << "CallWrapper";
253       break;
254     default:
255       llvm_unreachable("Invalid opcode");
256     }
257     dbgs() << ", seqno = " << SeqNo << ", tag-addr = " << TagAddr
258            << ", arg-buffer = " << formatv("{0:x}", ArgBytes.size())
259            << " bytes\n";
260   });
261   auto Err = T->sendMessage(OpC, SeqNo, TagAddr, ArgBytes);
262   LLVM_DEBUG({
263     if (Err)
264       dbgs() << "  \\--> SimpleRemoteEPC::sendMessage failed\n";
265   });
266   return Err;
267 }
268 
269 Error SimpleRemoteEPC::handleSetup(uint64_t SeqNo, ExecutorAddr TagAddr,
270                                    SimpleRemoteEPCArgBytesVector ArgBytes) {
271   if (SeqNo != 0)
272     return make_error<StringError>("Setup packet SeqNo not zero",
273                                    inconvertibleErrorCode());
274 
275   if (TagAddr)
276     return make_error<StringError>("Setup packet TagAddr not zero",
277                                    inconvertibleErrorCode());
278 
279   std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
280   auto I = PendingCallWrapperResults.find(0);
281   assert(PendingCallWrapperResults.size() == 1 &&
282          I != PendingCallWrapperResults.end() &&
283          "Setup message handler not connectly set up");
284   auto SetupMsgHandler = std::move(I->second);
285   PendingCallWrapperResults.erase(I);
286 
287   auto WFR =
288       shared::WrapperFunctionResult::copyFrom(ArgBytes.data(), ArgBytes.size());
289   SetupMsgHandler(std::move(WFR));
290   return Error::success();
291 }
292 
293 Error SimpleRemoteEPC::setup(Setup S) {
294   using namespace SimpleRemoteEPCDefaultBootstrapSymbolNames;
295 
296   std::promise<MSVCPExpected<SimpleRemoteEPCExecutorInfo>> EIP;
297   auto EIF = EIP.get_future();
298 
299   // Prepare a handler for the setup packet.
300   PendingCallWrapperResults[0] =
301     RunInPlace()(
302       [&](shared::WrapperFunctionResult SetupMsgBytes) {
303         if (const char *ErrMsg = SetupMsgBytes.getOutOfBandError()) {
304           EIP.set_value(
305               make_error<StringError>(ErrMsg, inconvertibleErrorCode()));
306           return;
307         }
308         using SPSSerialize =
309             shared::SPSArgList<shared::SPSSimpleRemoteEPCExecutorInfo>;
310         shared::SPSInputBuffer IB(SetupMsgBytes.data(), SetupMsgBytes.size());
311         SimpleRemoteEPCExecutorInfo EI;
312         if (SPSSerialize::deserialize(IB, EI))
313           EIP.set_value(EI);
314         else
315           EIP.set_value(make_error<StringError>(
316               "Could not deserialize setup message", inconvertibleErrorCode()));
317       });
318 
319   // Start the transport.
320   if (auto Err = T->start())
321     return Err;
322 
323   // Wait for setup packet to arrive.
324   auto EI = EIF.get();
325   if (!EI) {
326     T->disconnect();
327     return EI.takeError();
328   }
329 
330   LLVM_DEBUG({
331     dbgs() << "SimpleRemoteEPC received setup message:\n"
332            << "  Triple: " << EI->TargetTriple << "\n"
333            << "  Page size: " << EI->PageSize << "\n"
334            << "  Bootstrap map" << (EI->BootstrapMap.empty() ? " empty" : ":")
335            << "\n";
336     for (const auto &KV : EI->BootstrapMap)
337       dbgs() << "    " << KV.first() << ": " << KV.second.size()
338              << "-byte SPS encoded buffer\n";
339     dbgs() << "  Bootstrap symbols"
340            << (EI->BootstrapSymbols.empty() ? " empty" : ":") << "\n";
341     for (const auto &KV : EI->BootstrapSymbols)
342       dbgs() << "    " << KV.first() << ": " << KV.second << "\n";
343   });
344   TargetTriple = Triple(EI->TargetTriple);
345   PageSize = EI->PageSize;
346   BootstrapMap = std::move(EI->BootstrapMap);
347   BootstrapSymbols = std::move(EI->BootstrapSymbols);
348 
349   if (auto Err = getBootstrapSymbols(
350           {{JDI.JITDispatchContext, ExecutorSessionObjectName},
351            {JDI.JITDispatchFunction, DispatchFnName},
352            {RunAsMainAddr, rt::RunAsMainWrapperName},
353            {RunAsVoidFunctionAddr, rt::RunAsVoidFunctionWrapperName},
354            {RunAsIntFunctionAddr, rt::RunAsIntFunctionWrapperName}}))
355     return Err;
356 
357   if (auto DM =
358           EPCGenericDylibManager::CreateWithDefaultBootstrapSymbols(*this))
359     EPCDylibMgr = std::make_unique<EPCGenericDylibManager>(std::move(*DM));
360   else
361     return DM.takeError();
362 
363   // Set a default CreateMemoryManager if none is specified.
364   if (!S.CreateMemoryManager)
365     S.CreateMemoryManager = createDefaultMemoryManager;
366 
367   if (auto MemMgr = S.CreateMemoryManager(*this)) {
368     OwnedMemMgr = std::move(*MemMgr);
369     this->MemMgr = OwnedMemMgr.get();
370   } else
371     return MemMgr.takeError();
372 
373   // Set a default CreateMemoryAccess if none is specified.
374   if (!S.CreateMemoryAccess)
375     S.CreateMemoryAccess = createDefaultMemoryAccess;
376 
377   if (auto MemAccess = S.CreateMemoryAccess(*this)) {
378     OwnedMemAccess = std::move(*MemAccess);
379     this->MemAccess = OwnedMemAccess.get();
380   } else
381     return MemAccess.takeError();
382 
383   return Error::success();
384 }
385 
386 Error SimpleRemoteEPC::handleResult(uint64_t SeqNo, ExecutorAddr TagAddr,
387                                     SimpleRemoteEPCArgBytesVector ArgBytes) {
388   IncomingWFRHandler SendResult;
389 
390   if (TagAddr)
391     return make_error<StringError>("Unexpected TagAddr in result message",
392                                    inconvertibleErrorCode());
393 
394   {
395     std::lock_guard<std::mutex> Lock(SimpleRemoteEPCMutex);
396     auto I = PendingCallWrapperResults.find(SeqNo);
397     if (I == PendingCallWrapperResults.end())
398       return make_error<StringError>("No call for sequence number " +
399                                          Twine(SeqNo),
400                                      inconvertibleErrorCode());
401     SendResult = std::move(I->second);
402     PendingCallWrapperResults.erase(I);
403     releaseSeqNo(SeqNo);
404   }
405 
406   auto WFR =
407       shared::WrapperFunctionResult::copyFrom(ArgBytes.data(), ArgBytes.size());
408   SendResult(std::move(WFR));
409   return Error::success();
410 }
411 
412 void SimpleRemoteEPC::handleCallWrapper(
413     uint64_t RemoteSeqNo, ExecutorAddr TagAddr,
414     SimpleRemoteEPCArgBytesVector ArgBytes) {
415   assert(ES && "No ExecutionSession attached");
416   D->dispatch(makeGenericNamedTask(
417       [this, RemoteSeqNo, TagAddr, ArgBytes = std::move(ArgBytes)]() {
418         ES->runJITDispatchHandler(
419             [this, RemoteSeqNo](shared::WrapperFunctionResult WFR) {
420               if (auto Err =
421                       sendMessage(SimpleRemoteEPCOpcode::Result, RemoteSeqNo,
422                                   ExecutorAddr(), {WFR.data(), WFR.size()}))
423                 getExecutionSession().reportError(std::move(Err));
424             },
425             TagAddr, ArgBytes);
426       },
427       "callWrapper task"));
428 }
429 
430 Error SimpleRemoteEPC::handleHangup(SimpleRemoteEPCArgBytesVector ArgBytes) {
431   using namespace llvm::orc::shared;
432   auto WFR = WrapperFunctionResult::copyFrom(ArgBytes.data(), ArgBytes.size());
433   if (const char *ErrMsg = WFR.getOutOfBandError())
434     return make_error<StringError>(ErrMsg, inconvertibleErrorCode());
435 
436   detail::SPSSerializableError Info;
437   SPSInputBuffer IB(WFR.data(), WFR.size());
438   if (!SPSArgList<SPSError>::deserialize(IB, Info))
439     return make_error<StringError>("Could not deserialize hangup info",
440                                    inconvertibleErrorCode());
441   return fromSPSSerializable(std::move(Info));
442 }
443 
444 } // end namespace orc
445 } // end namespace llvm
446