[libc] Move RPC interface to libc/shared to export it
#117034
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Summary: Previous patches have made the `rpc.h` header independent of the `libc` internals. This allows us to include it directly rather than providing an indirect C API. This patch only does the work to move the header. A future patch will pull out the `rpc_server` interface and simply replace it with a single function that handles the opcodes.
jhuber6
requested review from
arsenm,
JonChesterfield,
lntue,
michaelrj-google and
shiltian
|
@llvm/pr-subscribers-libc Author: Joseph Huber (jhuber6) ChangesSummary: Patch is 41.55 KiB, truncated to 20.00 KiB below, full version: https://github.com/llvm/llvm-project/pull/117034.diff 9 Files Affected:
diff --git a/libc/src/__support/RPC/rpc.h b/libc/shared/rpc.h
similarity index 80%
rename from libc/src/__support/RPC/rpc.h
rename to libc/shared/rpc.h
index 30dd2c1a8125d7..489a8cebfb807c 100644
--- a/libc/src/__support/RPC/rpc.h
+++ b/libc/shared/rpc.h
@@ -15,16 +15,17 @@
//
//===----------------------------------------------------------------------===//
-#ifndef LLVM_LIBC_SRC___SUPPORT_RPC_RPC_H
-#define LLVM_LIBC_SRC___SUPPORT_RPC_RPC_H
+#ifndef LLVM_LIBC_SHARED_RPC_H
+#define LLVM_LIBC_SHARED_RPC_H
#include "rpc_util.h"
-#include "src/__support/macros/attributes.h"
-#include "src/__support/macros/config.h"
#include <stdint.h>
-namespace LIBC_NAMESPACE_DECL {
+#ifndef RPC_INLINE
+#define RPC_INLINE inline
+#endif
+
namespace rpc {
/// Use scoped atomic variants if they are available for the target.
@@ -70,12 +71,12 @@ constexpr static uint64_t MAX_PORT_COUNT = 4096;
/// - The server will always start with a 'recv' operation.
/// - Every 'send' or 'recv' call is mirrored by the other process.
template <bool Invert> struct Process {
- LIBC_INLINE Process() = default;
- LIBC_INLINE Process(const Process &) = delete;
- LIBC_INLINE Process &operator=(const Process &) = delete;
- LIBC_INLINE Process(Process &&) = default;
- LIBC_INLINE Process &operator=(Process &&) = default;
- LIBC_INLINE ~Process() = default;
+ RPC_INLINE Process() = default;
+ RPC_INLINE Process(const Process &) = delete;
+ RPC_INLINE Process &operator=(const Process &) = delete;
+ RPC_INLINE Process(Process &&) = default;
+ RPC_INLINE Process &operator=(Process &&) = default;
+ RPC_INLINE ~Process() = default;
uint32_t port_count = 0;
uint32_t *inbox = nullptr;
@@ -86,7 +87,7 @@ template <bool Invert> struct Process {
static constexpr uint64_t NUM_BITS_IN_WORD = sizeof(uint32_t) * 8;
uint32_t lock[MAX_PORT_COUNT / NUM_BITS_IN_WORD] = {0};
- LIBC_INLINE Process(uint32_t port_count, void *buffer)
+ RPC_INLINE Process(uint32_t port_count, void *buffer)
: port_count(port_count), inbox(reinterpret_cast<uint32_t *>(
advance(buffer, inbox_offset(port_count)))),
outbox(reinterpret_cast<uint32_t *>(
@@ -105,20 +106,20 @@ template <bool Invert> struct Process {
/// Header header[port_count];
/// Buffer packet[port_count][lane_size];
/// };
- LIBC_INLINE static constexpr uint64_t allocation_size(uint32_t port_count,
- uint32_t lane_size) {
+ RPC_INLINE static constexpr uint64_t allocation_size(uint32_t port_count,
+ uint32_t lane_size) {
return buffer_offset(port_count) + buffer_bytes(port_count, lane_size);
}
/// Retrieve the inbox state from memory shared between processes.
- LIBC_INLINE uint32_t load_inbox(uint64_t lane_mask, uint32_t index) const {
+ RPC_INLINE uint32_t load_inbox(uint64_t lane_mask, uint32_t index) const {
return rpc::broadcast_value(
lane_mask, __scoped_atomic_load_n(&inbox[index], __ATOMIC_RELAXED,
__MEMORY_SCOPE_SYSTEM));
}
/// Retrieve the outbox state from memory shared between processes.
- LIBC_INLINE uint32_t load_outbox(uint64_t lane_mask, uint32_t index) const {
+ RPC_INLINE uint32_t load_outbox(uint64_t lane_mask, uint32_t index) const {
return rpc::broadcast_value(
lane_mask, __scoped_atomic_load_n(&outbox[index], __ATOMIC_RELAXED,
__MEMORY_SCOPE_SYSTEM));
@@ -128,7 +129,7 @@ template <bool Invert> struct Process {
/// Equivalent to loading outbox followed by store of the inverted value
/// The outbox is write only by this warp and tracking the value locally is
/// cheaper than calling load_outbox to get the value to store.
- LIBC_INLINE uint32_t invert_outbox(uint32_t index, uint32_t current_outbox) {
+ RPC_INLINE uint32_t invert_outbox(uint32_t index, uint32_t current_outbox) {
uint32_t inverted_outbox = !current_outbox;
__scoped_atomic_thread_fence(__ATOMIC_RELEASE, __MEMORY_SCOPE_SYSTEM);
__scoped_atomic_store_n(&outbox[index], inverted_outbox, __ATOMIC_RELAXED,
@@ -138,8 +139,8 @@ template <bool Invert> struct Process {
// Given the current outbox and inbox values, wait until the inbox changes
// to indicate that this thread owns the buffer element.
- LIBC_INLINE void wait_for_ownership(uint64_t lane_mask, uint32_t index,
- uint32_t outbox, uint32_t in) {
+ RPC_INLINE void wait_for_ownership(uint64_t lane_mask, uint32_t index,
+ uint32_t outbox, uint32_t in) {
while (buffer_unavailable(in, outbox)) {
sleep_briefly();
in = load_inbox(lane_mask, index);
@@ -150,14 +151,14 @@ template <bool Invert> struct Process {
/// The packet is a linearly allocated array of buffers used to communicate
/// with the other process. This function returns the appropriate slot in this
/// array such that the process can operate on an entire warp or wavefront.
- LIBC_INLINE Buffer *get_packet(uint32_t index, uint32_t lane_size) {
+ RPC_INLINE Buffer *get_packet(uint32_t index, uint32_t lane_size) {
return &packet[index * lane_size];
}
/// Determines if this process needs to wait for ownership of the buffer. We
/// invert the condition on one of the processes to indicate that if one
/// process owns the buffer then the other does not.
- LIBC_INLINE static bool buffer_unavailable(uint32_t in, uint32_t out) {
+ RPC_INLINE static bool buffer_unavailable(uint32_t in, uint32_t out) {
bool cond = in != out;
return Invert ? !cond : cond;
}
@@ -166,7 +167,7 @@ template <bool Invert> struct Process {
/// lane_mask is a bitmap of the threads in the warp that would hold the
/// single lock on success, e.g. the result of rpc::get_lane_mask()
/// The lock is held when the n-th bit of the lock bitfield is set.
- LIBC_INLINE bool try_lock(uint64_t lane_mask, uint32_t index) {
+ RPC_INLINE bool try_lock(uint64_t lane_mask, uint32_t index) {
// On amdgpu, test and set to the nth lock bit and a sync_lane would suffice
// On volta, need to handle differences between the threads running and
// the threads that were detected in the previous call to get_lane_mask()
@@ -206,7 +207,7 @@ template <bool Invert> struct Process {
/// Unlock the lock at index. We need a lane sync to keep this function
/// convergent, otherwise the compiler will sink the store and deadlock.
- LIBC_INLINE void unlock(uint64_t lane_mask, uint32_t index) {
+ RPC_INLINE void unlock(uint64_t lane_mask, uint32_t index) {
// Do not move any writes past the unlock.
__scoped_atomic_thread_fence(__ATOMIC_RELEASE, __MEMORY_SCOPE_DEVICE);
@@ -219,40 +220,40 @@ template <bool Invert> struct Process {
}
/// Number of bytes to allocate for an inbox or outbox.
- LIBC_INLINE static constexpr uint64_t mailbox_bytes(uint32_t port_count) {
+ RPC_INLINE static constexpr uint64_t mailbox_bytes(uint32_t port_count) {
return port_count * sizeof(uint32_t);
}
/// Number of bytes to allocate for the buffer containing the packets.
- LIBC_INLINE static constexpr uint64_t buffer_bytes(uint32_t port_count,
- uint32_t lane_size) {
+ RPC_INLINE static constexpr uint64_t buffer_bytes(uint32_t port_count,
+ uint32_t lane_size) {
return port_count * lane_size * sizeof(Buffer);
}
/// Offset of the inbox in memory. This is the same as the outbox if inverted.
- LIBC_INLINE static constexpr uint64_t inbox_offset(uint32_t port_count) {
+ RPC_INLINE static constexpr uint64_t inbox_offset(uint32_t port_count) {
return Invert ? mailbox_bytes(port_count) : 0;
}
/// Offset of the outbox in memory. This is the same as the inbox if inverted.
- LIBC_INLINE static constexpr uint64_t outbox_offset(uint32_t port_count) {
+ RPC_INLINE static constexpr uint64_t outbox_offset(uint32_t port_count) {
return Invert ? 0 : mailbox_bytes(port_count);
}
/// Offset of the buffer containing the packets after the inbox and outbox.
- LIBC_INLINE static constexpr uint64_t header_offset(uint32_t port_count) {
+ RPC_INLINE static constexpr uint64_t header_offset(uint32_t port_count) {
return align_up(2 * mailbox_bytes(port_count), alignof(Header));
}
/// Offset of the buffer containing the packets after the inbox and outbox.
- LIBC_INLINE static constexpr uint64_t buffer_offset(uint32_t port_count) {
+ RPC_INLINE static constexpr uint64_t buffer_offset(uint32_t port_count) {
return align_up(header_offset(port_count) + port_count * sizeof(Header),
alignof(Buffer));
}
/// Conditionally set the n-th bit in the atomic bitfield.
- LIBC_INLINE static constexpr uint32_t set_nth(uint32_t *bits, uint32_t index,
- bool cond) {
+ RPC_INLINE static constexpr uint32_t set_nth(uint32_t *bits, uint32_t index,
+ bool cond) {
uint32_t slot = index / NUM_BITS_IN_WORD;
uint32_t bit = index % NUM_BITS_IN_WORD;
return __scoped_atomic_fetch_or(&bits[slot],
@@ -262,8 +263,8 @@ template <bool Invert> struct Process {
}
/// Conditionally clear the n-th bit in the atomic bitfield.
- LIBC_INLINE static constexpr uint32_t clear_nth(uint32_t *bits,
- uint32_t index, bool cond) {
+ RPC_INLINE static constexpr uint32_t clear_nth(uint32_t *bits, uint32_t index,
+ bool cond) {
uint32_t slot = index / NUM_BITS_IN_WORD;
uint32_t bit = index % NUM_BITS_IN_WORD;
return __scoped_atomic_fetch_and(&bits[slot],
@@ -275,8 +276,8 @@ template <bool Invert> struct Process {
/// Invokes a function accross every active buffer across the total lane size.
template <typename F>
-LIBC_INLINE static void invoke_rpc(F &&fn, uint32_t lane_size,
- uint64_t lane_mask, Buffer *slot) {
+RPC_INLINE static void invoke_rpc(F &&fn, uint32_t lane_size,
+ uint64_t lane_mask, Buffer *slot) {
if constexpr (is_process_gpu()) {
fn(&slot[rpc::get_lane_id()], rpc::get_lane_id());
} else {
@@ -290,40 +291,40 @@ LIBC_INLINE static void invoke_rpc(F &&fn, uint32_t lane_size,
/// processes. A port is conceptually an index into the memory provided by the
/// underlying process that is guarded by a lock bit.
template <bool T> struct Port {
- LIBC_INLINE Port(Process<T> &process, uint64_t lane_mask, uint32_t lane_size,
- uint32_t index, uint32_t out)
+ RPC_INLINE Port(Process<T> &process, uint64_t lane_mask, uint32_t lane_size,
+ uint32_t index, uint32_t out)
: process(process), lane_mask(lane_mask), lane_size(lane_size),
index(index), out(out), receive(false), owns_buffer(true) {}
- LIBC_INLINE ~Port() = default;
+ RPC_INLINE ~Port() = default;
private:
- LIBC_INLINE Port(const Port &) = delete;
- LIBC_INLINE Port &operator=(const Port &) = delete;
- LIBC_INLINE Port(Port &&) = default;
- LIBC_INLINE Port &operator=(Port &&) = default;
+ RPC_INLINE Port(const Port &) = delete;
+ RPC_INLINE Port &operator=(const Port &) = delete;
+ RPC_INLINE Port(Port &&) = default;
+ RPC_INLINE Port &operator=(Port &&) = default;
friend struct Client;
friend struct Server;
friend class rpc::optional<Port<T>>;
public:
- template <typename U> LIBC_INLINE void recv(U use);
- template <typename F> LIBC_INLINE void send(F fill);
+ template <typename U> RPC_INLINE void recv(U use);
+ template <typename F> RPC_INLINE void send(F fill);
template <typename F, typename U>
- LIBC_INLINE void send_and_recv(F fill, U use);
- template <typename W> LIBC_INLINE void recv_and_send(W work);
- LIBC_INLINE void send_n(const void *const *src, uint64_t *size);
- LIBC_INLINE void send_n(const void *src, uint64_t size);
+ RPC_INLINE void send_and_recv(F fill, U use);
+ template <typename W> RPC_INLINE void recv_and_send(W work);
+ RPC_INLINE void send_n(const void *const *src, uint64_t *size);
+ RPC_INLINE void send_n(const void *src, uint64_t size);
template <typename A>
- LIBC_INLINE void recv_n(void **dst, uint64_t *size, A &&alloc);
+ RPC_INLINE void recv_n(void **dst, uint64_t *size, A &&alloc);
- LIBC_INLINE uint32_t get_opcode() const {
+ RPC_INLINE uint32_t get_opcode() const {
return process.header[index].opcode;
}
- LIBC_INLINE uint32_t get_index() const { return index; }
+ RPC_INLINE uint32_t get_index() const { return index; }
- LIBC_INLINE void close() {
+ RPC_INLINE void close() {
// Wait for all lanes to finish using the port.
rpc::sync_lane(lane_mask);
@@ -346,16 +347,16 @@ template <bool T> struct Port {
/// The RPC client used to make requests to the server.
struct Client {
- LIBC_INLINE Client() = default;
- LIBC_INLINE Client(const Client &) = delete;
- LIBC_INLINE Client &operator=(const Client &) = delete;
- LIBC_INLINE ~Client() = default;
+ RPC_INLINE Client() = default;
+ RPC_INLINE Client(const Client &) = delete;
+ RPC_INLINE Client &operator=(const Client &) = delete;
+ RPC_INLINE ~Client() = default;
- LIBC_INLINE Client(uint32_t port_count, void *buffer)
+ RPC_INLINE Client(uint32_t port_count, void *buffer)
: process(port_count, buffer) {}
using Port = rpc::Port<false>;
- template <uint32_t opcode> LIBC_INLINE Port open();
+ template <uint32_t opcode> RPC_INLINE Port open();
private:
Process<false> process;
@@ -363,21 +364,21 @@ struct Client {
/// The RPC server used to respond to the client.
struct Server {
- LIBC_INLINE Server() = default;
- LIBC_INLINE Server(const Server &) = delete;
- LIBC_INLINE Server &operator=(const Server &) = delete;
- LIBC_INLINE ~Server() = default;
+ RPC_INLINE Server() = default;
+ RPC_INLINE Server(const Server &) = delete;
+ RPC_INLINE Server &operator=(const Server &) = delete;
+ RPC_INLINE ~Server() = default;
- LIBC_INLINE Server(uint32_t port_count, void *buffer)
+ RPC_INLINE Server(uint32_t port_count, void *buffer)
: process(port_count, buffer) {}
using Port = rpc::Port<true>;
- LIBC_INLINE rpc::optional<Port> try_open(uint32_t lane_size,
- uint32_t start = 0);
- LIBC_INLINE Port open(uint32_t lane_size);
+ RPC_INLINE rpc::optional<Port> try_open(uint32_t lane_size,
+ uint32_t start = 0);
+ RPC_INLINE Port open(uint32_t lane_size);
- LIBC_INLINE static uint64_t allocation_size(uint32_t lane_size,
- uint32_t port_count) {
+ RPC_INLINE static uint64_t allocation_size(uint32_t lane_size,
+ uint32_t port_count) {
return Process<true>::allocation_size(port_count, lane_size);
}
@@ -386,7 +387,7 @@ struct Server {
};
/// Applies \p fill to the shared buffer and initiates a send operation.
-template <bool T> template <typename F> LIBC_INLINE void Port<T>::send(F fill) {
+template <bool T> template <typename F> RPC_INLINE void Port<T>::send(F fill) {
uint32_t in = owns_buffer ? out ^ T : process.load_inbox(lane_mask, index);
// We need to wait until we own the buffer before sending.
@@ -401,7 +402,7 @@ template <bool T> template <typename F> LIBC_INLINE void Port<T>::send(F fill) {
}
/// Applies \p use to the shared buffer and acknowledges the send.
-template <bool T> template <typename U> LIBC_INLINE void Port<T>::recv(U use) {
+template <bool T> template <typename U> RPC_INLINE void Port<T>::recv(U use) {
// We only exchange ownership of the buffer during a receive if we are waiting
// for a previous receive to finish.
if (receive) {
@@ -424,7 +425,7 @@ template <bool T> template <typename U> LIBC_INLINE void Port<T>::recv(U use) {
/// Combines a send and receive into a single function.
template <bool T>
template <typename F, typename U>
-LIBC_INLINE void Port<T>::send_and_recv(F fill, U use) {
+RPC_INLINE void Port<T>::send_and_recv(F fill, U use) {
send(fill);
recv(use);
}
@@ -434,7 +435,7 @@ LIBC_INLINE void Port<T>::send_and_recv(F fill, U use) {
/// the copy back.
template <bool T>
template <typename W>
-LIBC_INLINE void Port<T>::recv_and_send(W work) {
+RPC_INLINE void Port<T>::recv_and_send(W work) {
recv(work);
send([](Buffer *, uint32_t) { /* no-op */ });
}
@@ -442,7 +443,7 @@ LIBC_INLINE void Port<T>::recv_and_send(W work) {
/// Helper routine to simplify the interface when sending from the GPU using
/// thread private pointers to the underlying value.
template <bool T>
-LIBC_INLINE void Port<T>::send_n(const void *src, uint64_t size) {
+RPC_INLINE void Port<T>::send_n(const void *src, uint64_t size) {
const void **src_ptr = &src;
uint64_t *size_ptr = &size;
send_n(src_ptr, size_ptr);
@@ -451,7 +452,7 @@ LIBC_INLINE void Port<T>::send_n(const void *src, uint64_t size) {
/// Sends an arbitrarily sized data buffer \p src across the shared channel in
/// multiples of the packet length.
template <bool T>
-LIBC_INLINE void Port<T>::send_n(const void *const *src, uint64_t *size) {
+RPC_INLINE void Port<T>::send_n(const void *const *src, uint64_t *size) {
uint64_t num_sends = 0;
send([&](Buffer *buffer, uint32_t id) {
reinterpret_cast<uint64_t *>(buffer->data)[0] = lane_value(size, id);
@@ -482,7 +483,7 @@ LIBC_INLINE void Port<T>::send_n(const void *const *src, uint64_t *size) {
/// size of the data so that we can initialize the size of the \p dst buffer.
template <bool T>
template <typename A>
-LIBC_INLINE void Port<T>::recv_n(void **dst, uint64_t *size, A &&alloc) {
+RPC_INLINE void Port<T>::recv_n(void **dst, uint64_t *size, A &&alloc) {
uint64_t num_recvs = 0;
recv([&](Buffer *buffer, uint32_t id) {
lane_value(size, id) = reinterpret_cast<uint64_t *>(buffer->data)[0];
@@ -516,7 +517,7 @@ LIBC_INLINE void Port<T>::recv_n(void **dst, uint64_t *size, A &&alloc) {
/// port. Each port instance uses an associated \p opcode to tell the server
/// what to do. The Client interface provides the appropriate lane size to the
/// port using the platform's returned value.
-template <uint32_t opcode> LIBC_INLINE Client::Port Client::open() {
+template <uint32_t opcode> RPC_INLINE Client::Port Client::open() {
// Repeatedly perform a naive linear scan for a port that can be opened to
// send data.
for (uint32_t index = 0;; ++index) {
@@ -550,7 +551,7 @@ template <uint32_t opcode> LIBC_INLINE Client::Port Client::open() {
/// Attempts to open a port to use as the server. The server can only open a
/// port if it has a pending receive operation
-LIBC_INLINE rpc::optional<typename Server::Port>
+RPC_INLINE rpc::optional<typename Server::Port>
Server::try_open(uint32_t lane_size, uint32_t start) {
// Perform a naive linear scan for a port that has a pending request.
for (uint32_t index = start; index < process.port_count; ++index) {
@@ -580,7 +581,7 @@ Server::try_open(uint32_t lane_size, uint32_t start) {
return rpc::nullopt;
}
-LIBC_INLINE Server::Port Server::open(uint32_t lane_size) {
+RPC_INLINE Server::Port Server::open(uint32_t lane_size) {
for (;;) {
if (rpc::optional<Server::Port> p = try_open(lane_size))
return rpc::move(p.value());
@@ -599,6 +600,5 @@ LIBC_INLINE Server::Port Server::open(uint32_t lane_size) {
#endif
} // namespace rpc
-} // namespace LIBC_NAMESPACE_DECL
-#endif
+#endif // LLVM_LIBC_SHARED_RPC_H
diff --git a/libc/src/__support/RPC/rpc_util.h b/libc/shared/rpc_util.h
similarity index 61%
rename from libc/src/__support/RPC/rpc_util.h
rename to libc/shared/rpc_util.h
index 7067dfc974eb31..502014d839ae94 100644
--- a/libc/src/__support/RPC/rpc_util.h
+++ b/libc/shared/rpc_util.h
@@ -6,11 +6,8 @@
//
//===----------------------------------------------------------------------===//
-#ifndef LLVM_LIBC_SRC___SUPPORT_RPC_RPC_UTIL_H
-#define LLVM_LIBC_SRC___SUPPORT_RPC_RPC_UTIL_H
-
-#include "src/__support/macros/attributes.h"
-#include "src/__support/macros/config.h"
+#ifndef LLVM_LIBC_SHARED_RPC_UTIL_H
+#define LLVM_LIBC_SHARED_RPC_UTIL_H
#inc...
[truncated]
|
|
For future reference, discussed with @michaelrj-google and the current plan is to store this in |
michaelrj-google
approved these changes
Nov 22, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary:
Previous patches have made the
rpc.hheader independent of thelibcinternals. This allows us to include it directly rather than providing
an indirect C API. This patch only does the work to move the header. A
future patch will pull out the
rpc_serverinterface and simply replaceit with a single function that handles the opcodes.