Subject: Relacy verification: Asymmetric SMR / Hazard Pointer with mo_consume and zero seq_cst...
Just wanted to share a quick update on a Safe Memory Reclamation (SMR)
hazard pointer engine I've been putting through its paces using Relacy.
I managed to design an asymmetric, poller-driven deferred reclamation
model that successfully eliminates heavy mo_seq_cst operations entirely
from the runtime path.
Reader Fast Path: Uses pure relaxed pointer loads and hazard stores,
combined with a memory_order_consume validation load (node1 == node0) to cleanly chain the data dependency without heavy ordering bottlenecks.
Writer Exchanges: Uses mo_acq_rel instead of sequential consistency.
Poller Mechanism: (ah-hoc) Handles cleanup safely via an exclusive mutex (m_poll), a central deferred queue (m_defer0), a two-pass hazard scan (prv_poll), and a rl::systemwide_fence($) between passes to prevent any use-after-free conditions. It just cleared a rigorous simulation run
of 500,000 iterations cleanly across 5 readers and 3 writers with zero
data races, leaks, or race conditions
I attached my code. Can you run it on your end? Any thoughts? Thanks everybody!
struct ct::smr_test::smr_test_v000
13% (65536/500000)
26% (131072/500000)
39% (196608/500000)
52% (262144/500000)
65% (327680/500000)
78% (393216/500000)
91% (458752/500000)
iterations: 500000
total time: 14093
throughput: 35478
D:\ct_dev\ct_msvc\projects\ct_relacy_test\x64\Release\ct_relacy_test.exe (process 2408) exited with code 0 (0x0).
Press any key to close this window . . .
my experimental test code:
ct_smr_test.hpp
______________________
#pragma once
#include <relacy/relacy.hpp>
#include <cstdio>
#include <cstddef>
#include <cstdlib>
#include <array>
#include <string>
#include <vector>
namespace ct
{
namespace smr_test
{
constexpr unsigned long CT_THREADS_READERS = 5;
constexpr unsigned long CT_THREADS_WRITERS = 3;
constexpr unsigned long CT_THREADS = CT_THREADS_READERS + CT_THREADS_WRITERS;
constexpr unsigned long CT_THREADS_READER_ITER_N = 8;
constexpr unsigned long CT_THREADS_WRITER_ITER_N = 8;
struct per_node
{
VAR_T(per_node*) m_defer = { nullptr };
VAR_T(unsigned long) m_data = { 0 };
};
struct per_thread
{
rl::atomic<per_node*> m_haz_0 = { nullptr };
};
struct poller
{
rl::mutex m_poll;
VAR_T(per_node*) m_defer0 = nullptr;
void
prv_poll(
per_node*& defer,
std::array<per_thread, CT_THREADS>& threads
) {
per_node* head = defer;
defer = nullptr; // Take ownership of the caller's
local batch
// Merge any previously deferred nodes from m_defer0
into the processing batch
per_node* curr = VAR(m_defer0);
VAR(m_defer0) = nullptr;
while (curr)
{
per_node* next_node = VAR(curr->m_defer);
VAR(curr->m_defer) = head;
head = curr;
curr = next_node;
}
// Two-Pass Hazard Pointer Scan:
// Pass 1: Capture all thread hazard slots
std::vector<per_node*> haz_pass1(threads.size());
for (size_t i = 0; i < threads.size(); ++i)
{
haz_pass1[i] =
threads[i].m_haz_0.load(rl::mo_relaxed, $);
}
// Synchronization barrier between scans
rl::systemwide_fence($);
// Pass 2: Capture all thread hazard slots again
std::vector<per_node*> haz_pass2(threads.size());
for (size_t i = 0; i < threads.size(); ++i)
{
haz_pass2[i] =
threads[i].m_haz_0.load(rl::mo_relaxed, $);
}
per_node* remaining = nullptr;
while (head)
{
per_node* next = VAR(head->m_defer);
bool protected_node = false;
// Check if the node is protected in either pass
snapshot
for (auto haz : haz_pass1)
{
if (haz == head)
{
protected_node = true;
break;
}
}
if (!protected_node)
{
for (auto haz : haz_pass2)
{
if (haz == head)
{
protected_node = true;
break;
}
}
}
if (protected_node)
{
// Keep it in the deferred list
VAR(head->m_defer) = remaining;
remaining = head;
}
else
{
// Safe to reclaim!
delete head;
}
head = next;
}
// Store any still-protected nodes back into the
central poller queue
VAR(m_defer0) = remaining;
}
bool
try_poll(
per_node*& defer,
std::array<per_thread, CT_THREADS>& threads
) {
if (m_poll.try_lock($))
{
prv_poll(defer, threads);
m_poll.unlock($);
return true;
}
return false;
}
void
poll(
per_node*& defer,
std::array<per_thread, CT_THREADS>& threads
) {
m_poll.lock($);
prv_poll(defer, threads);
m_poll.unlock($);
}
};
struct state
{
std::array<per_thread, CT_THREADS> m_threads;
poller m_poller = { };
rl::atomic<per_node*> m_node = { };
// relacy hooks, so to speak... ;^o
// relaxed because everything either has not
// started yet, or everything is already done...
void
before()
{
per_node* node = new per_node();
VAR(node->m_data) = 456;
VAR(node->m_defer) = nullptr;
m_node.store(node, rl::mo_relaxed, $);
}
void
after()
{
per_node* node = m_node.load(rl::mo_relaxed, $);
if (node)
{
delete node;
}
// Also clean up any lingering nodes left in the
poller's m_defer0 at teardown
per_node* poll_defer = VAR(m_poller.m_defer0);
while (poll_defer)
{
per_node* next = VAR(poll_defer->m_defer);
delete poll_defer;
poll_defer = next;
}
}
per_node*
smr_load(
unsigned long tidx
) {
for (;;)
{
per_node* node0 = m_node.load(rl::mo_relaxed, $);
if (! node0)
{
RL_ASSERT(0);
break;
}
m_threads[tidx].m_haz_0.store(node0,
rl::mo_relaxed, $);
// actually, iirc, a single seq_cst right here
should work?
per_node* node1 = m_node.load(rl::mo_consume, $);
if (node0 == node1)
{
return node0;
}
}
return nullptr;
}
void
smr_unload(
unsigned long tidx
) {
// humm...
m_threads[tidx].m_haz_0.store(nullptr, rl::mo_relaxed, $);
}
};
// the actual test unit...
struct smr_test_v000
: rl::test_suite<smr_test_v000, CT_THREADS>
{
state m_state = { };
void before()
{
m_state.before();
}
void after()
{
m_state.after();
}
void
thread_reader(
unsigned int tidx
) {
for (unsigned long i = 0; i < CT_THREADS_READER_ITER_N;
++i)
{
per_node* node = m_state.smr_load(tidx);
if (node)
{
unsigned long data = VAR(node->m_data);
RL_ASSERT(data == 123 || data == 456);
m_state.smr_unload(tidx);
}
}
}
void
thread_writer(
unsigned int tidx
) {
per_node* my_defer = nullptr;
for (unsigned long i = 0; i < CT_THREADS_WRITER_ITER_N;
++i)
{
per_node* node = new per_node();
VAR(node->m_data) = 456;
VAR(node->m_defer) = nullptr;
per_node* old = m_state.m_node.exchange(node, rl::mo_acq_rel, $);
if (old)
{
VAR(old->m_defer) = my_defer;
my_defer = old;
}
// Opportunistically try to poll and sweep nodes
m_state.m_poller.try_poll(my_defer, m_state.m_threads);
}
// Drain remaining deferred nodes upon thread exit
m_state.m_poller.poll(my_defer, m_state.m_threads);
}
void
thread(
unsigned int tidx
) {
if (tidx < CT_THREADS_READERS)
{
thread_reader(tidx);
}
else
{
thread_writer(tidx);
}
}
};
}
}
namespace ct
{
namespace smr_test
{
namespace test
{
inline void
manifest(
rl::test_params& p
) {
rl::simulate<ct::smr_test::smr_test_v000>(p);
}
}
}
}
________________________
ct_main.hpp
________________________
#include <relacy/relacy.hpp>
// tests
#include "ct_smr_test.hpp"
#include "ct_multex_test.hpp"
#include <cstdio>
#include <cstddef>
#include <cstdlib>
#include <string>
#include <vector>
int main(int argc, char** argv)
{
{
rl::test_params p;
p.iteration_count = argc > 1 ? (unsigned)atoi(argv[1]) : 500000;
p.execution_depth_limit = 100000;
if (argc > 2) {
std::string m = argv[2];
if (m == "full") p.search_type = rl::sched_full;
else if (m == "bound") { p.search_type = rl::sched_bound; p.context_bound = argc > 3 ? atoi(argv[3]) : 2; }
}
{
//ct::multex_test::test::manifest(p);
ct::smr_test::test::manifest(p);
}
}
return 0;
}
/*
#include <relacy/relacy.hpp>
#include <iostream>
int
main()
{
std::cout << "ct_relacy_test\n";
std::cout << "____________________________________________" <<
std::endl;
{
}
std::cout << "____________________________________________" <<
std::endl;
std::cout << "ct_relacy_test fin! :^)\n";
return 0;
}
*/
________________________
--- PyGate Linux v1.5.19
* Origin: Dragon's Lair, PyGate NNTP<>Fido Gate (3:633/10)