/* Simple Proxy Collector (DWCAS) - Relacy Version
http://www.1024cores.net/home/relacy-race-detector
Copyright 3/27/2018
___________________________________________________*/
#include <relacy/relacy_std.hpp>
#include <cstdio>
#include <climits>
/* Membar Abstraction
___________________________________________________*/
#define ct_mb_relaxed std::memory_order_relaxed
#define ct_mb_acquire std::memory_order_acquire
#define ct_mb_release std::memory_order_release
#define ct_mb_seq_cst std::memory_order_seq_cst
#define ct_mb_fence(mb_membar) std::atomic_thread_fence(mb_membar)
/* Simple Proxy Collector C++11
___________________________________________________*/
namespace ct {
namespace proxy {
// User object base
struct object
{
virtual ~object() {}
};
// Proxy node
struct node
{
std::atomic<std::intptr_t> count;
VAR_T(node*) next;
VAR_T(object*) obj;
node(std::intptr_t count_, node* next_, object* obj_)
: count(count_), next(next_), obj(obj_)
{
}
~node()
{
}
};
// DWCAS target
struct anchor
{
std::intptr_t count;
node* head;
anchor()
{
}
anchor(std::intptr_t count_, node* head_)
{
count = count_;
head = head_;
}
bool operator == (anchor const& right) const
{
return count == right.count && head == right.head;
}
anchor operator + (anchor const& right) const
{
anchor res;
res.count = count + right.count;
res.head = right.head;
return res;
}
anchor operator - (anchor const& right) const
{
anchor res;
res.count = count - right.count;
res.head = right.head;
return res;
}
};
std::ostream& operator << (std::ostream& s, anchor const& right)
{
return s << "{" << right.count << "," << right.head << "}";
}
// Collector Logic
struct gc
{
std::atomic<anchor> head;
gc() : head(anchor{ 0, new node(0, nullptr, nullptr) }) {}
~gc()
{
anchor cmp = head.load(ct_mb_relaxed);
RL_ASSERT(cmp.count > -1);
RL_ASSERT(VAR(cmp.head->next) == nullptr);
prv_dtor(cmp.head);
}
// Drops a reference count on a node
bool prv_release(node* n)
{
std::intptr_t count = n->count.fetch_sub(2, ct_mb_relaxed);
if (count == 3) return true;
return false;
}
// Deletes a node
void prv_dtor(node* n)
{
object* obj = VAR(n->obj);
if (obj != nullptr) delete obj;
delete n;
}
// Dumps backlinked nodes
void prv_dump(node* n)
{
node* cur = VAR(n->next);
VAR(n->next) = nullptr;
// Release and collect in cur LIFO
while (prv_release(cur))
{
ct_mb_fence(ct_mb_acquire);
node* next = VAR(cur->next);
VAR(cur->next) = n;
n = cur;
cur = next;
}
// Destroy cur LIFO
while (n)
{
node* next = VAR(n->next);
prv_dtor(n);
n = next;
}
}
// Collects a node
void prv_collect(node* n)
{
anchor xchg = { 0, n };
ct_mb_fence(ct_mb_release);
// Swap in new node
anchor cmp = head.exchange(xchg, ct_mb_relaxed);
// Backlink
ct_mb_fence(ct_mb_acquire);
VAR(cmp.head->next) = xchg.head;
ct_mb_fence(ct_mb_release);
// Release and mark as collected
std::intptr_t count = cmp.head->count.fetch_add(cmp.count + 1, ct_mb_relaxed);
if (count + cmp.count == 0)
{
prv_dump(cmp.head);
}
}
// Acquires current node
node* acquire()
{
anchor cmp = head.load(ct_mb_relaxed);
anchor xchg = { cmp.count + 2, cmp.head };
while (! head.compare_exchange_weak(cmp, xchg, ct_mb_relaxed))
{
xchg = { cmp.count + 2, cmp.head };
}
ct_mb_fence(ct_mb_acquire);
return cmp.head;
}
// Release a node
void release(node* n)
{
ct_mb_fence(ct_mb_release);
if (prv_release(n))
{
ct_mb_fence(ct_mb_acquire);
prv_dump(n);
}
}
// Collects an object
void collect(object* obj)
{
prv_collect(new node(2, nullptr, obj));
}
};
}}
// Test object
struct foo : public ct::proxy::object
{
std::atomic<foo*> next;
foo(foo* next_ = nullptr) : next(next_) { }
virtual ~foo()
{
//std::cout << "foo::~foo()\n";
}
};
// An atomic LIFO of test objects
struct foo_lifo
{
std::atomic<foo*> head;
foo_lifo(foo* next_ = nullptr) : head(next_) { }
void push(foo* n)
{
foo* cmp = head.load(ct_mb_relaxed);
do {
n->next.store(cmp, ct_mb_relaxed);
ct_mb_fence(ct_mb_release);
} while (! head.compare_exchange_weak(cmp, n, ct_mb_relaxed));
}
foo* flush()
{
foo* cmp = head.exchange(nullptr, ct_mb_relaxed);
if (cmp) ct_mb_fence(ct_mb_acquire);
return cmp;
}
};
// Program Logic
#define READERS 5
#define WRITERS 3
#define ITERS 5
#define THREADS (READERS + WRITERS)
struct ct_proxy_test
: rl::test_suite<ct_proxy_test, THREADS>
{
ct::proxy::gc g_gc;
foo_lifo g_lifo;
void before()
{
//std::printf("before()\n");
}
void after()
{
//std::printf("after()\n");
}
// Reader threads...
void thread_reader(unsigned int tidx)
{
//std::printf("thread_reader::%u\n", tidx);
ct::proxy::node* pcn = g_gc.acquire();
{
foo* cur = g_lifo.head.load(ct_mb_relaxed);
ct_mb_fence(ct_mb_acquire);
while (cur)
{
foo* next = cur->next.load(ct_mb_relaxed);
rl::yield(1, $);
cur = next;
}
}
g_gc.release(pcn);
}
// Writer threads...
void thread_writer(unsigned int tidx)
{
//std::printf("thread_writer::%u\n", tidx);
for (unsigned int i = 0; i < ITERS; ++i)
{
// Add some nodes, and peform some collections...
g_lifo.push(new foo());
g_lifo.push(new foo());
if (! (i % 2)) g_gc.collect(nullptr);
rl::yield(1, $);
g_lifo.push(new foo());
if (! (i % 3)) g_lifo.push(new foo());
// Flush
foo* cur = g_lifo.flush();
// Collect the foo's... ;^)
while (cur)
{
foo* next = cur->next.load(ct_mb_relaxed);
rl::yield(1, $);
g_gc.collect(cur);
cur = next;
}
}
}
// Entry...
void thread(unsigned int tidx)
{
if (tidx < READERS)
{
thread_reader(tidx);
}
else
{
thread_writer(tidx);
}
}
};
int main()
{
{
rl::test_params p;
p.iteration_count = 12345600; // for existing proxy gc code
//p.execution_depth_limit = 3333;
//p.search_type = rl::random_scheduler_type;
//p.search_type = rl::sched_bound;
//p.search_type = rl::fair_full_search_scheduler_type;
//p.search_type = rl::fair_context_bound_scheduler_type;
rl::simulate<ct_proxy_test>(p);
}
std::puts("\nTest Complete!\n");
std::fflush(stdout);
std::getchar();
return 0;
}
_______________________
Thanks.