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
73SimpleRemoteEPCTransportClient::~SimpleRemoteEPCTransportClient() = default;
74SimpleRemoteEPCTransport::~SimpleRemoteEPCTransport() = default;
75
76Expected<std::unique_ptr<FDSimpleRemoteEPCTransport>>
77FDSimpleRemoteEPCTransport::Create(SimpleRemoteEPCTransportClient &C, int InFD,
78 int OutFD) {
79#if LLVM_ENABLE_THREADS
80 if (InFD == -1)
81 return make_error<StringError>(Args: "Invalid input file descriptor " +
82 Twine(InFD),
83 Args: inconvertibleErrorCode());
84 if (OutFD == -1)
85 return make_error<StringError>(Args: "Invalid output file descriptor " +
86 Twine(OutFD),
87 Args: inconvertibleErrorCode());
88 std::unique_ptr<FDSimpleRemoteEPCTransport> FDT(
89 new FDSimpleRemoteEPCTransport(C, InFD, OutFD));
90 return std::move(FDT);
91#else
92 return make_error<StringError>("FD-based SimpleRemoteEPC transport requires "
93 "thread support, but llvm was built with "
94 "LLVM_ENABLE_THREADS=Off",
95 inconvertibleErrorCode());
96#endif
97}
98
99FDSimpleRemoteEPCTransport::~FDSimpleRemoteEPCTransport() {
100#if LLVM_ENABLE_THREADS
101 ListenerThread.join();
102#endif
103}
104
105Error FDSimpleRemoteEPCTransport::start() {
106#if LLVM_ENABLE_THREADS
107 ListenerThread = std::thread([this]() { listenLoop(); });
108 return Error::success();
109#endif
110 llvm_unreachable("Should not be called with LLVM_ENABLE_THREADS=Off");
111}
112
113Error FDSimpleRemoteEPCTransport::sendMessage(SimpleRemoteEPCOpcode OpC,
114 uint64_t SeqNo,
115 ExecutorAddr TagAddr,
116 ArrayRef<char> ArgBytes) {
117 char HeaderBuffer[FDMsgHeader::Size];
118
119 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::MsgSizeOffset)) =
120 FDMsgHeader::Size + ArgBytes.size();
121 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::OpCOffset)) =
122 static_cast<uint64_t>(OpC);
123 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::SeqNoOffset)) = SeqNo;
124 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::TagAddrOffset)) =
125 TagAddr.getValue();
126
127 std::lock_guard<std::mutex> Lock(M);
128 if (Disconnected)
129 return make_error<StringError>(Args: "FD-transport disconnected",
130 Args: inconvertibleErrorCode());
131 if (int ErrNo = writeBytes(Src: HeaderBuffer, Size: FDMsgHeader::Size))
132 return errorCodeToError(EC: std::error_code(ErrNo, std::generic_category()));
133 if (int ErrNo = writeBytes(Src: ArgBytes.data(), Size: ArgBytes.size()))
134 return errorCodeToError(EC: std::error_code(ErrNo, std::generic_category()));
135 return Error::success();
136}
137
138void FDSimpleRemoteEPCTransport::disconnect() {
139 if (Disconnected)
140 return; // Return if already disconnected.
141
142 Disconnected = true;
143 bool CloseOutFD = InFD != OutFD;
144
145#ifndef _WIN32
146 // We need to shutdown the socket to wake up (and terminate) any ongoing
147 // blocking read on this FD. If the FD is not a socket, shutdown will just
148 // complain through errno (instead of crashing).
149 // FIXME: what about Windows?
150 ::shutdown(fd: InFD, how: CloseOutFD ? SHUT_RD : SHUT_RDWR);
151#endif
152 // Close InFD.
153 while (close(fd: InFD) == -1) {
154 if (errno == EBADF)
155 break;
156 }
157
158 // Close OutFD.
159 if (CloseOutFD) {
160#ifndef _WIN32
161 // FIXME: what about Windows?
162 ::shutdown(fd: OutFD, SHUT_WR);
163#endif
164 while (close(fd: OutFD) == -1) {
165 if (errno == EBADF)
166 break;
167 }
168 }
169}
170
171static Error makeUnexpectedEOFError() {
172 return make_error<StringError>(Args: "Unexpected end-of-file",
173 Args: inconvertibleErrorCode());
174}
175
176Error FDSimpleRemoteEPCTransport::readBytes(char *Dst, size_t Size,
177 bool *IsEOF) {
178 assert((Size == 0 || Dst) && "Attempt to read into null.");
179 ssize_t Completed = 0;
180 while (Completed < static_cast<ssize_t>(Size)) {
181 ssize_t Read = ::read(fd: InFD, buf: Dst + Completed, nbytes: Size - Completed);
182 if (Read <= 0) {
183 auto ErrNo = errno;
184 if (Read == 0) {
185 if (Completed == 0 && IsEOF) {
186 *IsEOF = true;
187 return Error::success();
188 } else
189 return makeUnexpectedEOFError();
190 } else if (ErrNo == EAGAIN || ErrNo == EINTR)
191 continue;
192 else {
193 std::lock_guard<std::mutex> Lock(M);
194 if (Disconnected && IsEOF) { // disconnect called, pretend this is EOF.
195 *IsEOF = true;
196 return Error::success();
197 }
198 return errorCodeToError(
199 EC: std::error_code(ErrNo, std::generic_category()));
200 }
201 }
202 Completed += Read;
203 }
204 return Error::success();
205}
206
207int FDSimpleRemoteEPCTransport::writeBytes(const char *Src, size_t Size) {
208 assert((Size == 0 || Src) && "Attempt to append from null.");
209 ssize_t Completed = 0;
210 while (Completed < static_cast<ssize_t>(Size)) {
211 ssize_t Written = ::write(fd: OutFD, buf: Src + Completed, n: Size - Completed);
212 if (Written < 0) {
213 auto ErrNo = errno;
214 if (ErrNo == EAGAIN || ErrNo == EINTR)
215 continue;
216 else
217 return ErrNo;
218 }
219 Completed += Written;
220 }
221 return 0;
222}
223
224void FDSimpleRemoteEPCTransport::listenLoop() {
225 Error Err = Error::success();
226 do {
227
228 char HeaderBuffer[FDMsgHeader::Size];
229 // Read the header buffer.
230 {
231 bool IsEOF = false;
232 if (auto Err2 = readBytes(Dst: HeaderBuffer, Size: FDMsgHeader::Size, IsEOF: &IsEOF)) {
233 Err = joinErrors(E1: std::move(Err), E2: std::move(Err2));
234 break;
235 }
236 if (IsEOF)
237 break;
238 }
239
240 // Decode header buffer.
241 uint64_t MsgSize;
242 SimpleRemoteEPCOpcode OpC;
243 uint64_t SeqNo;
244 ExecutorAddr TagAddr;
245
246 MsgSize =
247 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::MsgSizeOffset));
248 OpC = static_cast<SimpleRemoteEPCOpcode>(static_cast<uint64_t>(
249 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::OpCOffset))));
250 SeqNo =
251 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::SeqNoOffset));
252 TagAddr.setValue(
253 *((support::ulittle64_t *)(HeaderBuffer + FDMsgHeader::TagAddrOffset)));
254
255 if (MsgSize < FDMsgHeader::Size) {
256 Err = joinErrors(E1: std::move(Err),
257 E2: make_error<StringError>(Args: "Message size too small",
258 Args: inconvertibleErrorCode()));
259 break;
260 }
261
262 // Read the argument bytes.
263 auto ArgBytes =
264 shared::WrapperFunctionBuffer::allocate(Size: MsgSize - FDMsgHeader::Size);
265 if (auto Err2 = readBytes(Dst: ArgBytes.data(), Size: ArgBytes.size())) {
266 Err = joinErrors(E1: std::move(Err), E2: std::move(Err2));
267 break;
268 }
269
270 if (auto Action =
271 C.handleMessage(OpC, SeqNo, TagAddr, ArgBytes: std::move(ArgBytes))) {
272 if (*Action == SimpleRemoteEPCTransportClient::EndSession)
273 break;
274 } else {
275 Err = joinErrors(E1: std::move(Err), E2: Action.takeError());
276 break;
277 }
278 } while (true);
279
280 // Attempt to close FDs, set Disconnected to true so that subsequent
281 // sendMessage calls fail.
282 disconnect();
283
284 // Call up to the client to handle the disconnection.
285 C.handleDisconnect(Err: std::move(Err));
286}
287
288} // end namespace orc
289} // end namespace llvm
290