1//===------ SimpleRemoteEPCUtils.cpp - Utils for Simple Remote EPC --------===//
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// Message definitions and other utilities for SimpleRemoteEPC and
10// SimpleRemoteEPCServer.
11//
12//===----------------------------------------------------------------------===//
13
14#include "llvm/ExecutionEngine/Orc/Shared/SimpleRemoteEPCUtils.h"
15#include "llvm/Config/llvm-config.h" // for LLVM_ENABLE_THREADS
16#include "llvm/Support/Endian.h"
17
18#if !defined(_MSC_VER) && !defined(__MINGW32__)
19#include <unistd.h>
20#else
21#include <io.h>
22#endif
23#ifndef _WIN32
24#include <sys/socket.h>
25#endif
26
27namespace {
28
29struct FDMsgHeader {
30 static constexpr unsigned MsgSizeOffset = 0;
31 static constexpr unsigned OpCOffset = MsgSizeOffset + sizeof(uint64_t);
32 static constexpr unsigned SeqNoOffset = OpCOffset + sizeof(uint64_t);
33 static constexpr unsigned TagAddrOffset = SeqNoOffset + sizeof(uint64_t);
34 static constexpr unsigned Size = TagAddrOffset + sizeof(uint64_t);
35};
36
37} // namespace
38
39namespace llvm {
40namespace orc {
41namespace SimpleRemoteEPCDefaultBootstrapSymbolNames {
42
43const char *ExecutorSessionObjectName =
44 "__llvm_orc_SimpleRemoteEPC_dispatch_ctx";
45const char *DispatchFnName = "__llvm_orc_SimpleRemoteEPC_dispatch_fn";
46
47} // end namespace SimpleRemoteEPCDefaultBootstrapSymbolNames
48
49shared::WrapperFunctionBuffer encodeHangupPayload(Error Err) {
50 using SPSSerialize = shared::SPSArgList<shared::SPSError>;
51 auto SE = shared::detail::toSPSSerializable(Err: std::move(Err));
52 auto Payload =
53 shared::WrapperFunctionBuffer::allocate(Size: SPSSerialize::size(Arg: SE));
54 shared::SPSOutputBuffer OB(Payload.data(), Payload.size());
55 bool Success = SPSSerialize::serialize(OB, Arg: SE);
56 (void)Success;
57 assert(Success && "Hangup payload serialization should not fail");
58 return Payload;
59}
60
61Error decodeHangupPayload(shared::WrapperFunctionBuffer Payload) {
62 assert(!Payload.getOutOfBandError() &&
63 "Hangup payload should not be an out-of-band error buffer");
64
65 shared::detail::SPSSerializableError Info;
66 shared::SPSInputBuffer IB(Payload.data(), Payload.size());
67 if (!shared::SPSArgList<shared::SPSError>::deserialize(IB, Arg&: Info))
68 return make_error<StringError>(Args: "Could not deserialize hangup info",
69 Args: inconvertibleErrorCode());
70 return shared::detail::fromSPSSerializable(BSE: std::move(Info));
71}
72
73std::pair<ExecutorAddr, shared::WrapperFunctionBuffer>
74encodeResultMessage(shared::WrapperFunctionBuffer ResultBytes) {
75 auto Tag = [](SimpleRemoteEPCResultKind K) {
76 return ExecutorAddr(static_cast<uint64_t>(K));
77 };
78
79 const char *ErrMsg = ResultBytes.getOutOfBandError();
80 if (!ErrMsg)
81 return {Tag(SimpleRemoteEPCResultKind::Value), std::move(ResultBytes)};
82
83 using SPSSerialize = shared::SPSArgList<shared::SPSString>;
84 StringRef M(ErrMsg);
85 auto Payload = shared::WrapperFunctionBuffer::allocate(Size: SPSSerialize::size(Arg: M));
86 shared::SPSOutputBuffer OB(Payload.data(), Payload.size());
87 bool Success = SPSSerialize::serialize(OB, Arg: M);
88 (void)Success;
89 assert(Success && "Out-of-band error serialization should not fail");
90 return {Tag(SimpleRemoteEPCResultKind::OutOfBandError), std::move(Payload)};
91}
92
93Expected<shared::WrapperFunctionBuffer>
94decodeResultMessage(ExecutorAddr TagAddr,
95 shared::WrapperFunctionBuffer Payload) {
96 using UT = std::underlying_type_t<SimpleRemoteEPCResultKind>;
97 UT KindVal = TagAddr.getValue();
98 if (KindVal > static_cast<UT>(SimpleRemoteEPCResultKind::LastResultKind))
99 return make_error<StringError>(Args: "Unexpected result kind " + Twine(KindVal) +
100 " in result message",
101 Args: inconvertibleErrorCode());
102
103 switch (static_cast<SimpleRemoteEPCResultKind>(KindVal)) {
104 case SimpleRemoteEPCResultKind::Value:
105 return std::move(Payload);
106 case SimpleRemoteEPCResultKind::OutOfBandError: {
107 // A malformed payload is reported as the out-of-band error itself: the
108 // call waiting on this result must be unblocked either way, and an error
109 // about the error is more use to the caller than a dead session.
110 std::string Msg;
111 shared::SPSInputBuffer IB(Payload.data(), Payload.size());
112 if (!shared::SPSArgList<shared::SPSString>::deserialize(IB, Arg&: Msg))
113 return shared::WrapperFunctionBuffer::createOutOfBandError(
114 Msg: "Could not deserialize out-of-band error message");
115 return shared::WrapperFunctionBuffer::createOutOfBandError(Msg);
116 }
117 }
118 llvm_unreachable("Invalid result kind");
119}
120
121SimpleRemoteEPCTransportClient::~SimpleRemoteEPCTransportClient() = default;
122SimpleRemoteEPCTransport::~SimpleRemoteEPCTransport() = default;
123
124Expected<std::unique_ptr<FDSimpleRemoteEPCTransport>>
125FDSimpleRemoteEPCTransport::Create(SimpleRemoteEPCTransportClient &C, int InFD,
126 int OutFD) {
127#if LLVM_ENABLE_THREADS
128 if (InFD == -1)
129 return make_error<StringError>(Args: "Invalid input file descriptor " +
130 Twine(InFD),
131 Args: inconvertibleErrorCode());
132 if (OutFD == -1)
133 return make_error<StringError>(Args: "Invalid output file descriptor " +
134 Twine(OutFD),
135 Args: inconvertibleErrorCode());
136 std::unique_ptr<FDSimpleRemoteEPCTransport> FDT(
137 new FDSimpleRemoteEPCTransport(C, InFD, OutFD));
138 return std::move(FDT);
139#else
140 return make_error<StringError>("FD-based SimpleRemoteEPC transport requires "
141 "thread support, but llvm was built with "
142 "LLVM_ENABLE_THREADS=Off",
143 inconvertibleErrorCode());
144#endif
145}
146
147FDSimpleRemoteEPCTransport::~FDSimpleRemoteEPCTransport() {
148#if LLVM_ENABLE_THREADS
149 ListenerThread.join();
150#endif
151}
152
153Error FDSimpleRemoteEPCTransport::start() {
154#if LLVM_ENABLE_THREADS
155 ListenerThread = std::thread([this]() { listenLoop(); });
156 return Error::success();
157#endif
158 llvm_unreachable("Should not be called with LLVM_ENABLE_THREADS=Off");
159}
160
161Error FDSimpleRemoteEPCTransport::sendMessage(SimpleRemoteEPCOpcode OpC,
162 uint64_t SeqNo,
163 ExecutorAddr TagAddr,
164 ArrayRef<char> ArgBytes) {
165 char HeaderBuffer[FDMsgHeader::Size];
166
167 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::MsgSizeOffset)) =
168 FDMsgHeader::Size + ArgBytes.size();
169 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::OpCOffset)) =
170 static_cast<uint64_t>(OpC);
171 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::SeqNoOffset)) = SeqNo;
172 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::TagAddrOffset)) =
173 TagAddr.getValue();
174
175 std::lock_guard<std::mutex> Lock(M);
176 if (Disconnected)
177 return make_error<StringError>(Args: "FD-transport disconnected",
178 Args: inconvertibleErrorCode());
179 if (int ErrNo = writeBytes(Src: HeaderBuffer, Size: FDMsgHeader::Size))
180 return errorCodeToError(EC: std::error_code(ErrNo, std::generic_category()));
181 if (int ErrNo = writeBytes(Src: ArgBytes.data(), Size: ArgBytes.size()))
182 return errorCodeToError(EC: std::error_code(ErrNo, std::generic_category()));
183 return Error::success();
184}
185
186void FDSimpleRemoteEPCTransport::disconnect() {
187 if (Disconnected)
188 return; // Return if already disconnected.
189
190 Disconnected = true;
191 bool CloseOutFD = InFD != OutFD;
192
193#ifndef _WIN32
194 // We need to shutdown the socket to wake up (and terminate) any ongoing
195 // blocking read on this FD. If the FD is not a socket, shutdown will just
196 // complain through errno (instead of crashing).
197 // FIXME: what about Windows?
198 ::shutdown(fd: InFD, how: CloseOutFD ? SHUT_RD : SHUT_RDWR);
199#endif
200 // Close InFD.
201 while (close(fd: InFD) == -1) {
202 if (errno == EBADF)
203 break;
204 }
205
206 // Close OutFD.
207 if (CloseOutFD) {
208#ifndef _WIN32
209 // FIXME: what about Windows?
210 ::shutdown(fd: OutFD, SHUT_WR);
211#endif
212 while (close(fd: OutFD) == -1) {
213 if (errno == EBADF)
214 break;
215 }
216 }
217}
218
219static Error makeUnexpectedEOFError() {
220 return make_error<StringError>(Args: "Unexpected end-of-file",
221 Args: inconvertibleErrorCode());
222}
223
224Error FDSimpleRemoteEPCTransport::readBytes(char *Dst, size_t Size,
225 bool *IsEOF) {
226 assert((Size == 0 || Dst) && "Attempt to read into null.");
227 ssize_t Completed = 0;
228 while (Completed < static_cast<ssize_t>(Size)) {
229 ssize_t Read = ::read(fd: InFD, buf: Dst + Completed, nbytes: Size - Completed);
230 if (Read <= 0) {
231 auto ErrNo = errno;
232 if (Read == 0) {
233 if (Completed == 0 && IsEOF) {
234 *IsEOF = true;
235 return Error::success();
236 } else
237 return makeUnexpectedEOFError();
238 } else if (ErrNo == EAGAIN || ErrNo == EINTR)
239 continue;
240 else {
241 std::lock_guard<std::mutex> Lock(M);
242 if (Disconnected && IsEOF) { // disconnect called, pretend this is EOF.
243 *IsEOF = true;
244 return Error::success();
245 }
246 return errorCodeToError(
247 EC: std::error_code(ErrNo, std::generic_category()));
248 }
249 }
250 Completed += Read;
251 }
252 return Error::success();
253}
254
255int FDSimpleRemoteEPCTransport::writeBytes(const char *Src, size_t Size) {
256 assert((Size == 0 || Src) && "Attempt to append from null.");
257 ssize_t Completed = 0;
258 while (Completed < static_cast<ssize_t>(Size)) {
259 ssize_t Written = ::write(fd: OutFD, buf: Src + Completed, n: Size - Completed);
260 if (Written < 0) {
261 auto ErrNo = errno;
262 if (ErrNo == EAGAIN || ErrNo == EINTR)
263 continue;
264 else
265 return ErrNo;
266 }
267 Completed += Written;
268 }
269 return 0;
270}
271
272void FDSimpleRemoteEPCTransport::listenLoop() {
273 Error Err = Error::success();
274 do {
275
276 char HeaderBuffer[FDMsgHeader::Size];
277 // Read the header buffer.
278 {
279 bool IsEOF = false;
280 if (auto Err2 = readBytes(Dst: HeaderBuffer, Size: FDMsgHeader::Size, IsEOF: &IsEOF)) {
281 Err = joinErrors(E1: std::move(Err), E2: std::move(Err2));
282 break;
283 }
284 if (IsEOF)
285 break;
286 }
287
288 // Decode header buffer.
289 uint64_t MsgSize;
290 SimpleRemoteEPCOpcode OpC;
291 uint64_t SeqNo;
292 ExecutorAddr TagAddr;
293
294 MsgSize =
295 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::MsgSizeOffset));
296 OpC = static_cast<SimpleRemoteEPCOpcode>(static_cast<uint64_t>(
297 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::OpCOffset))));
298 SeqNo =
299 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::SeqNoOffset));
300 TagAddr.setValue(
301 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::TagAddrOffset)));
302
303 if (MsgSize < FDMsgHeader::Size) {
304 Err = joinErrors(E1: std::move(Err),
305 E2: make_error<StringError>(Args: "Message size too small",
306 Args: inconvertibleErrorCode()));
307 break;
308 }
309
310 // Read the argument bytes.
311 auto ArgBytes =
312 shared::WrapperFunctionBuffer::allocate(Size: MsgSize - FDMsgHeader::Size);
313 if (auto Err2 = readBytes(Dst: ArgBytes.data(), Size: ArgBytes.size())) {
314 Err = joinErrors(E1: std::move(Err), E2: std::move(Err2));
315 break;
316 }
317
318 if (auto Action =
319 C.handleMessage(OpC, SeqNo, TagAddr, ArgBytes: std::move(ArgBytes))) {
320 if (*Action == SimpleRemoteEPCTransportClient::EndSession)
321 break;
322 } else {
323 Err = joinErrors(E1: std::move(Err), E2: Action.takeError());
324 break;
325 }
326 } while (true);
327
328 // Attempt to close FDs, set Disconnected to true so that subsequent
329 // sendMessage calls fail.
330 disconnect();
331
332 // Call up to the client to handle the disconnection.
333 C.handleDisconnect(Err: std::move(Err));
334}
335
336} // end namespace orc
337} // end namespace llvm
338