Skip to content

Simplify shm_unix SharedMemoryRegistry #7030

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
niklasf wants to merge 1 commit into from
Closed

Simplify shm_unix SharedMemoryRegistry #7030

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
180 changes: 46 additions & 134 deletions src/shm_unix.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,60 +56,37 @@ namespace Stockfish::shm {

namespace detail {

class SharedMemoryBase {
public:
enum class CloseType {
Normal,
AtExit,
};

virtual ~SharedMemoryBase() = default;
virtual void close(CloseType) noexcept = 0;
virtual const std::string& name() const noexcept = 0;
};

class SharedMemoryRegistry {
private:
static std::mutex registry_mutex_;
static std::vector<SharedMemoryBase*> active_instances_;
inline static std::once_flag register_atexit_once_;
inline static std::mutex socket_paths_mutex_;
inline static std::vector<std::string> socket_paths_;

public:
static void register_instance(SharedMemoryBase* instance) {
std::scoped_lock lock(registry_mutex_);
active_instances_.push_back(instance);
static void register_socket_path(std::string path) {
std::call_once(register_atexit_once_,
[]() { std::atexit(SharedMemoryRegistry::cleanup_at_exit); });

std::scoped_lock lock(socket_paths_mutex_);
socket_paths_.push_back(std::move(path));
}

static void unregister_instance(SharedMemoryBase* instance) {
std::scoped_lock lock(registry_mutex_);
active_instances_.erase(
std::remove(active_instances_.begin(), active_instances_.end(), instance),
active_instances_.end());
static void unlink_socket_path(const std::string& path) {
std::scoped_lock lock(socket_paths_mutex_);
unlink(path.c_str());
socket_paths_.erase(std::remove(socket_paths_.begin(), socket_paths_.end(), path),
socket_paths_.end());
}

static void cleanup_at_exit() noexcept {
std::scoped_lock lock(registry_mutex_);
std::scoped_lock lock(socket_paths_mutex_);
// TODO: we litter .sock files upon an abnormal exit (e.g. Ctrl-C)
// but it's unsafe to acquire the lock in a signal handler
for (auto* instance : active_instances_)
instance->close(SharedMemoryBase::CloseType::AtExit);
for (auto const& path : socket_paths_)
unlink(path.c_str());
}
};

inline std::mutex SharedMemoryRegistry::registry_mutex_;
inline std::vector<SharedMemoryBase*> SharedMemoryRegistry::active_instances_;

class CleanupHooks {
private:
static std::once_flag register_once_;

static void register_atexit() noexcept { std::atexit(SharedMemoryRegistry::cleanup_at_exit); }

public:
static void ensure_registered() noexcept { std::call_once(register_once_, register_atexit); }
};

inline std::once_flag CleanupHooks::register_once_;

} // namespace detail


Expand Down Expand Up @@ -211,7 +188,7 @@ struct InitLock {
};

template<typename T>
class SharedMemory: public detail::SharedMemoryBase {
class SharedMemory {
static_assert(std::is_trivially_copyable_v<T>, "T must be trivially copyable");
static_assert(!std::is_pointer_v<T>, "T cannot be a pointer type");

Expand All @@ -229,9 +206,9 @@ class SharedMemory: public detail::SharedMemoryBase {
std::string init_lock_path_;

// serve requests for the shared segment on this .sock
std::string socket_path_;
std::optional<std::thread> server_thread_;
UniqueFd shutdown_; // close to signal server thread shutdown
std::string socket_path_;
std::thread server_thread_;
UniqueFd shutdown_; // close to signal server thread shutdown

static std::string make_sentinel_base(const std::string& name) {
char buf[32];
Expand All @@ -246,81 +223,35 @@ class SharedMemory: public detail::SharedMemoryBase {
name_(name),
shared_dir_(tempRoot.prefix + "/" + make_sentinel_base(name)),
init_lock_path_(shared_dir_ + "/init_lock"),
socket_path_(shared_dir_ + "/" + std::to_string(getpid()) + ".sock"),
server_thread_(std::nullopt) {}
socket_path_(shared_dir_ + "/" + std::to_string(getpid()) + ".sock") {}

~SharedMemory() noexcept override {
detail::SharedMemoryRegistry::unregister_instance(this);
SharedMemory::close(CloseType::Normal);
~SharedMemory() noexcept {
if (!socket_path_.empty())
detail::SharedMemoryRegistry::unlink_socket_path(socket_path_);

shutdown_.reset();
if (server_thread_.joinable())
server_thread_.join();

if (mapped_ptr_)
munmap(mapped_ptr_, sizeof(T));
}

SharedMemory(const SharedMemory&) = delete;
SharedMemory& operator=(const SharedMemory&) = delete;

SharedMemory(SharedMemory&& other) noexcept :
name_(std::move(other.name_)),
mapped_ptr_(other.mapped_ptr_),
data_ptr_(other.data_ptr_),
shared_dir_(std::move(other.shared_dir_)),
init_lock_path_(std::move(other.init_lock_path_)),
socket_path_(std::move(other.socket_path_)),
server_thread_(std::move(other.server_thread_)),
shutdown_(std::move(other.shutdown_)) {

detail::SharedMemoryRegistry::unregister_instance(&other);
detail::SharedMemoryRegistry::register_instance(this);

other.mapped_ptr_ = nullptr;
other.data_ptr_ = nullptr;
}
SharedMemory(SharedMemory&& other) noexcept { swap(other); }

SharedMemory& operator=(SharedMemory&& other) noexcept {
if (this != &other)
{
detail::SharedMemoryRegistry::unregister_instance(this);
close(CloseType::Normal);

name_ = std::move(other.name_);
mapped_ptr_ = other.mapped_ptr_;
data_ptr_ = other.data_ptr_;
shared_dir_ = std::move(other.shared_dir_);
init_lock_path_ = std::move(other.init_lock_path_);
socket_path_ = std::move(other.socket_path_);
server_thread_ = std::move(other.server_thread_);
shutdown_ = std::move(other.shutdown_);

detail::SharedMemoryRegistry::unregister_instance(&other);
detail::SharedMemoryRegistry::register_instance(this);

other.mapped_ptr_ = nullptr;
other.data_ptr_ = nullptr;
}
swap(other);
return *this;
}

void close(CloseType closeType) noexcept override {
if (closeType == CloseType::AtExit)
{
// Don't unmap on exit as this may cause currently searching threads to segfault.
// Also, don't join() the server thread on exit.
if (server_thread_)
{
server_thread_->detach();
server_thread_ = std::nullopt;
}
}
else
{
unmap_region();
}
reset();
}

bool is_mapped() const noexcept { return mapped_ptr_ != nullptr; }

bool is_serving() const noexcept { return server_thread_.has_value(); }
bool is_serving() const noexcept { return server_thread_.joinable(); }

const std::string& name() const noexcept override { return name_; }
const std::string& name() const noexcept { return name_; }

const T& get() const noexcept {
assert(data_ptr_ != nullptr);
Expand All @@ -329,31 +260,15 @@ class SharedMemory: public detail::SharedMemoryBase {
}

private:
void reset() noexcept {
if (!socket_path_.empty())
{
unlink(socket_path_.c_str());
}

shutdown_.reset();

if (server_thread_ && server_thread_->joinable())
{
server_thread_->join();
server_thread_ = std::nullopt;
}

mapped_ptr_ = nullptr;
data_ptr_ = nullptr;
}

void unmap_region() noexcept {
if (mapped_ptr_)
{
munmap(mapped_ptr_, sizeof(T));
mapped_ptr_ = nullptr;
data_ptr_ = nullptr;
}
void swap(SharedMemory& other) noexcept {
std::swap(name_, other.name_);
std::swap(mapped_ptr_, other.mapped_ptr_);
std::swap(data_ptr_, other.data_ptr_);
std::swap(shared_dir_, other.shared_dir_);
std::swap(init_lock_path_, other.init_lock_path_);
std::swap(socket_path_, other.socket_path_);
std::swap(server_thread_, other.server_thread_);
std::swap(shutdown_, other.shutdown_);
}

// Discover all peers in the shared dir
Expand Down Expand Up @@ -539,8 +454,6 @@ class SharedMemory: public detail::SharedMemoryBase {

public:
[[nodiscard]] bool open(const T& initial_value) noexcept {
detail::CleanupHooks::ensure_registered();

if (socket_path_.size() >= sizeof(sockaddr_un::sun_path))
return false;

Expand Down Expand Up @@ -606,8 +519,6 @@ class SharedMemory: public detail::SharedMemoryBase {

mapped_ptr_ = data_ptr_ = mapped_mem;

detail::SharedMemoryRegistry::register_instance(this); // register for cleanup at exit

int shutdown_pipe[2];
#if !defined(__APPLE__)
if (pipe2(shutdown_pipe, O_CLOEXEC) != 0)
Expand All @@ -629,6 +540,7 @@ class SharedMemory: public detail::SharedMemoryBase {
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, socket_path_.c_str(), sizeof(addr.sun_path) - 1);

detail::SharedMemoryRegistry::register_socket_path(socket_path_);
unlink(socket_path_.c_str());
if (bind(server_fd.get(), reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) == -1
|| listen(server_fd.get(), 5) == -1)
Expand Down
Loading