296 lines · cpp
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-exception6//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_THREADS32void 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, [this]() { return Outstanding == 0; });53}54#endif55 56StringMap<ExecutorAddr> SimpleRemoteEPCServer::defaultBootstrapSymbols() {57 StringMap<ExecutorAddr> DBS;58 rt_bootstrap::addTo(DBS);59 return DBS;60}61 62Expected<SimpleRemoteEPCTransportClient::HandleMessageAction>63SimpleRemoteEPCServer::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 = " << TagAddr89 << ", 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>("Unexpected opcode",96 inconvertibleErrorCode());97 98 // TODO: Clean detach message?99 switch (OpC) {100 case SimpleRemoteEPCOpcode::Setup:101 return make_error<StringError>("Unexpected Setup opcode",102 inconvertibleErrorCode());103 case SimpleRemoteEPCOpcode::Hangup:104 return SimpleRemoteEPCTransportClient::EndSession;105 case SimpleRemoteEPCOpcode::Result:106 if (auto Err = handleResult(SeqNo, TagAddr, std::move(ArgBytes)))107 return std::move(Err);108 break;109 case SimpleRemoteEPCOpcode::CallWrapper:110 handleCallWrapper(SeqNo, TagAddr, 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, [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(TmpPending, 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("disconnecting"));135 136 // Wait for dispatcher to clear.137 D->shutdown();138 139 // Shut down services.140 while (!Services.empty()) {141 ShutdownErr =142 joinErrors(std::move(ShutdownErr), Services.back()->shutdown());143 Services.pop_back();144 }145 146 std::lock_guard<std::mutex> Lock(ServerStateMutex);147 ShutdownErr = joinErrors(std::move(ShutdownErr), 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 = " << TagAddr178 << ", 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 else200 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(this);209 EI.BootstrapSymbols[DispatchFnName] = ExecutorAddr::fromPtr(jitDispatchEntry);210 addDefaultBootstrapValuesForHostProcess(EI.BootstrapMap, EI.BootstrapSymbols);211 212 using SPSSerialize =213 shared::SPSArgList<shared::SPSSimpleRemoteEPCExecutorInfo>;214 auto SetupPacketBytes =215 shared::WrapperFunctionResult::allocate(SPSSerialize::size(EI));216 shared::SPSOutputBuffer OB(SetupPacketBytes.data(), SetupPacketBytes.size());217 if (!SPSSerialize::serialize(OB, EI))218 return make_error<StringError>("Could not send setup packet",219 inconvertibleErrorCode());220 221 return sendMessage(SimpleRemoteEPCOpcode::Setup, 0, ExecutorAddr(),222 {SetupPacketBytes.data(), SetupPacketBytes.size()});223}224 225Error SimpleRemoteEPCServer::handleResult(226 uint64_t SeqNo, ExecutorAddr TagAddr,227 SimpleRemoteEPCArgBytesVector ArgBytes) {228 std::promise<shared::WrapperFunctionResult> *P = nullptr;229 {230 std::lock_guard<std::mutex> Lock(ServerStateMutex);231 auto I = PendingJITDispatchResults.find(SeqNo);232 if (I == PendingJITDispatchResults.end())233 return make_error<StringError>("No call for sequence number " +234 Twine(SeqNo),235 inconvertibleErrorCode());236 P = I->second;237 PendingJITDispatchResults.erase(I);238 releaseSeqNo(SeqNo);239 }240 auto R = shared::WrapperFunctionResult::allocate(ArgBytes.size());241 memcpy(R.data(), ArgBytes.data(), 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 SimpleRemoteEPCArgBytesVector ArgBytes) {249 D->dispatch([this, RemoteSeqNo, TagAddr, ArgBytes = std::move(ArgBytes)]() {250 using WrapperFnTy =251 shared::CWrapperFunctionResult (*)(const char *, size_t);252 auto *Fn = TagAddr.toPtr<WrapperFnTy>();253 shared::WrapperFunctionResult ResultBytes(254 Fn(ArgBytes.data(), ArgBytes.size()));255 if (auto Err = sendMessage(SimpleRemoteEPCOpcode::Result, RemoteSeqNo,256 ExecutorAddr(),257 {ResultBytes.data(), ResultBytes.size()}))258 ReportError(std::move(Err));259 });260}261 262shared::WrapperFunctionResult263SimpleRemoteEPCServer::doJITDispatch(const void *FnTag, const char *ArgData,264 size_t ArgSize) {265 uint64_t SeqNo;266 std::promise<shared::WrapperFunctionResult> ResultP;267 auto ResultF = ResultP.get_future();268 {269 std::lock_guard<std::mutex> Lock(ServerStateMutex);270 if (RunState != ServerRunning)271 return shared::WrapperFunctionResult::createOutOfBandError(272 "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(SimpleRemoteEPCOpcode::CallWrapper, SeqNo,280 ExecutorAddr::fromPtr(FnTag), {ArgData, ArgSize}))281 ReportError(std::move(Err));282 283 return ResultF.get();284}285 286shared::CWrapperFunctionResult287SimpleRemoteEPCServer::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 orc295} // end namespace llvm296