c++ wrapper for marked pointers

adf12bd3 · Nickolai Zeldovich · 5fd96933 · adf12bd3 · adf12bd3 · adf12bd3
--- a/cpprt.cc
+++ b/cpprt.cc
@@ -17,6 +17,12 @@ operator new(unsigned long nbytes, void *buf)
  return buf;
 }
+void *
+operator new[](unsigned long nbytes)
+{
+  return kmalloc(nbytes);
+}
 void
 operator delete(void *p)
 {

--- a/crange.cc
+++ b/crange.cc
@@ -50,10 +50,6 @@ extern "C" {
 // maybe many epochs later.
 //
-#define MARKED(x) (((uptr) (x)) & 0x1)
-#define MARK(x) ((struct range *) (((uptr) (x)) | 0x1))
-#define WOMARK(x) ((struct range *) (((uptr) (x)) & ~0x1))
 enum { crange_debug = 0 };
 enum { crange_checking = 0 };
@@ -84,9 +80,9 @@ range_draw_nlevel(int nlevel)
 void range::print(int l)
 {
-  cprintf ("0x%lx-0x%lx(%lu) 0x%lx, c %d, t %d, n 0x%lx m 0x%lx\n",
+  cprintf ("0x%lx-0x%lx(%lu) 0x%lx, c %d, t %d, n 0x%lx m %d\n",
           key, key+size, size, (long) value, curlevel.load(), nlevel,
-           (long) next, MARKED(next[l]));
+           (long) next, (bool) next[l].mark());
 }
 range::~range()
@@ -96,7 +92,7 @@ range::~range()
  this->cr->check(this);
  //    assert(this->curlevel == -1);
  for (int l = 0; l < this->nlevel; l++) {
-    this->next[l] = (struct range *) 0xDEADBEEF;
+    next[l] = (struct range *) 0xDEADBEEF;
  }
  kmalignfree(this->lock);
  kmfree(this->next);
@@ -119,7 +115,7 @@ void range::dec_ref(void)
  }
 }
-range::range(crange *cr, u64 k, u64 sz, void *v, struct range *n, int nlevel = 0)
+range::range(crange *cr, u64 k, u64 sz, void *v, markptr<range> n, int nlevel)
  : rcu_freed("range_delayed")
 {
  if (crange_debug)
@@ -132,7 +128,7 @@ range::range(crange *cr, u64 k, u64 sz, void *v, struct range *n, int nlevel = 0
  this->curlevel = 0;
  if (nlevel == 0) this->nlevel = range_draw_nlevel(cr->nlevel);
  else this->nlevel = nlevel;
-  this->next = (struct range**) kmalloc(sizeof(sizeof(this->next[0])) * this->nlevel); // cache align?
+  this->next = new markptr<range>[this->nlevel]; // cache align?
  assert(this->next);
  this->next[0] = n;
  for (int l = 1; l < this->nlevel; l++) this->next[l] = 0;
@@ -147,10 +143,10 @@ range::range(crange *cr, u64 k, u64 sz, void *v, struct range *n, int nlevel = 0
 static range *insert(struct range *head, struct range *r)
 {
-  struct range *n, *p;
+  markptr<range> n, p;
  p = nullptr;
  for (n = head; n != 0; p = n, n = n->next[0]) {
-    assert(!MARKED(n));
+    assert(!n.mark());
    if (r->key < n->key) {
      break;
    }
@@ -166,9 +162,9 @@ static range *insert(struct range *head, struct range *r)
 }
 // lock p if this->next == e and p isn't marked for deletion.  if not, return failure.
-int range::lockif(range *e)
+int range::lockif(markptr<range> e)
 {
-  assert(!MARKED(e));
+  assert(!e.mark());
  acquire(this->lock);
  if (this->next[0] == e) {
      return 1;
@@ -187,11 +183,10 @@ int range::lockif(range *e)
 static void mark(range *f, range *s)
 {
  struct range *e;
-  for (e = f; e && e != s; e = WOMARK(e->next[0])) {
+  for (e = f; e && e != s; e = e->next[0].ptr()) {
    assert(e);
-    for (int l = e->nlevel-1; l >= 0; l--) {
+    for (int l = e->nlevel-1; l >= 0; l--)
-        (void) __sync_fetch_and_or(&(e->next[l]), (struct range *)0x1);
+      e->next[l].mark() = 1;
-    }
  }
 }
@@ -199,7 +194,7 @@ static void mark(range *f, range *s)
 static void unlockn(range *f, range *l)
 {
  struct range *e;
-  for (e = f; e != l; e = WOMARK(e->next[0])) {
+  for (e = f; e != l; e = e->next[0].ptr()) {
    assert(e);
    release(e->lock);
  }
@@ -210,7 +205,7 @@ static void unlockn(range *f, range *l)
 static void freen(struct range *f, struct range *l)
 {
  struct range *e;
-  for (e = f; e != l; e = WOMARK(e->next[0])) {
+  for (e = f; e != l; e = e->next[0].ptr()) {
    assert(e);
    assert(e->curlevel >= 0);
    e->dec_ref();
@@ -276,11 +271,10 @@ static range *replace(u64 k, u64 sz, void *v, range *f, range *l, range *s)
 void
 crange::print(int full)
 {
-  struct range *e;
  for (int l = 0; l < this->nlevel; l++) {
    int c = 0;
    cprintf("crange %d: ", l);
-    for (e = this->crange_head->next[l]; e; e = WOMARK(e->next[l])) {
+    for (range *e = crange_head->next[l].ptr(); e != 0; e = e->next[l].ptr()) {
      c++;
      if (full) e->print(l);
    }
@@ -300,8 +294,8 @@ crange::~crange()
 {
  if (crange_debug) cprintf("crange_free: 0x%lx\n", (u64) this);
  range *e, *n;
-  for (e = WOMARK(this->crange_head->next[0]); e; e = n) {
+  for (e = this->crange_head->next[0].ptr(); e; e = n) {
-    n = WOMARK(e->next[0]);
+    n = e->next[0].ptr();
    delete e;
  }
  delete this->crange_head;
@@ -315,30 +309,30 @@ void crange::check(struct range *absent)
  int t = mycpu()->id;
  struct range *e, *s;
  for (int l = 0; l < this->nlevel; l++) {
-    for (e = this->crange_head->next[l]; e; e = s) {
+    for (e = this->crange_head->next[l].ptr(); e; e = s) {
      assert(e->curlevel < this->nlevel);
      if (absent == e)  {
 	cprintf("%d: check level failed; 0x%lx is present\n", l, (u64) absent);
 	assert(0);
      }
      // look for e level down, but only for non-marked nodes.
-      if (l > 0 && e->next[l] != 0 && !MARKED(e->next[l])) {
+      if (l > 0 && e->next[l] != 0 && !e->next[l].mark()) {
 	struct range *n;
-	for (n = WOMARK(this->crange_head->next[l-1]); n && n != e; n = WOMARK(n->next[l-1]))
+	for (n = crange_head->next[l-1].ptr(); n && n != e; n = n->next[l-1].ptr())
 	  ;
 	__sync_synchronize();
 	// if e is marked now, skip the check (the memory barrier ensures that we reread it
 	// from memory (and not from a register)
-	if (!MARKED(e->next[l]) && n != e) {
+	if (!e->next[l].mark() && n != e) {
-	  cprintf("%d: check level %d failed 0x%lx-0x%lx(%lu) m %lu c %d t %d; in high level but not low\n", t, l, e->key, e->key+e->size, e->size, MARKED(e->next[l]), e->curlevel.load(), e->nlevel);
+	  cprintf("%d: check level %d failed 0x%lx-0x%lx(%lu) m %d c %d t %d; in high level but not low\n", t, l, e->key, e->key+e->size, e->size, (bool)e->next[l].mark(), e->curlevel.load(), e->nlevel);
 	  this->print(1);
 	  assert(0);
 	}
      }
      // check if succ range is after n
-      s = WOMARK(e->next[l]);
+      s = e->next[l].ptr();
      assert(s != e);
-      if (!MARKED(e->next[l]) && s && (e->key + e->size > s->key)) {
+      if (!e->next[l].mark() && s && (e->key + e->size > s->key)) {
 	if (crange_debug) cprintf("%d: e(%lu,%lu) overlaps with s(%lu,%lu)\n", t, e->key, e->size, s->key, e->size);
 	this->print(1);
 	assert(0);
@@ -354,11 +348,11 @@ int crange::del_index(range *p0, range **e, int l)
 {
  int r = 1;
  assert(l < (*e)->nlevel);
-  if (!MARKED((*e)->next[l]))  // don't remove unmarked ranges from index
+  if (!(*e)->next[l].mark())  // don't remove unmarked ranges from index
    return r;
  if (l == 0) return 0;    // but not on level 0; they are locked when removed
  // crange_check(cr, NULL);
-  while (*e && MARKED((*e)->next[l])) {
+  while (*e && (*e)->next[l].mark()) {
 #if 0
    if (l != (*e)->curlevel) {
      // range is still in the index one level up, back out.  we want to remove it first 
@@ -368,11 +362,11 @@ int crange::del_index(range *p0, range **e, int l)
      goto done;
    }
 #endif
-    int cas = __sync_bool_compare_and_swap(&(p0->next[l]), *e, WOMARK((*e)->next[l]));
+    bool cas = p0->next[l].cmpxch(*e, (*e)->next[l].ptr().load());
    if (cas) {
      assert((*e)->curlevel >= 0);
      (*e)->dec_ref();
-      *e = WOMARK((*e)->next[l]);
+      *e = (*e)->next[l].ptr();
    } else {  
      // cprintf("%d: crange_del_index: retry del %u(%u)\n", mycpu()->id, (*e)->key, (*e)->key + (*e)->size);
      r = -1;
@@ -387,18 +381,18 @@ int crange::del_index(range *p0, range **e, int l)
 // Insert e into index one level up, between p and s, if e hasn't been inserted
 // yet on that level.
-void crange::add_index(int l, range *e, range *p1, range *s1)
+void crange::add_index(int l, range *e, range *p1, markptr<range> s1)
 {
  if (l >= e->nlevel-1) return;
-  if (MARKED(e->next[l+1])) return;
+  if (e->next[l+1].mark()) return;
  // crange_check(cr, NULL);
  if (cmpxch(&e->curlevel, l, l+1)) {
    assert(e->curlevel < e->nlevel);
    // this is the core inserting at level l+1, but some core may be deleting
-    struct range *s = WOMARK(s1);
+    struct range *s = s1.ptr();
    do {
-      struct range *n = e->next[l+1];   // Null and perhaps marked
+      markptr<range> n = e->next[l+1];   // Null and perhaps marked
-      if (MARKED(n)) {
+      if (n.mark()) {
 	// this range has been deleted, don't insert into index.
 	// undo increment of cur->level.
 	e->curlevel--;
@@ -406,9 +400,9 @@ void crange::add_index(int l, range *e, range *p1, range *s1)
 	goto done;
      }
      assert (n == 0);
-    } while (!__sync_bool_compare_and_swap(&(e->next[l+1]), 0, s));
+    } while (!e->next[l+1].cmpxch(0, s));
-    if (!__sync_bool_compare_and_swap(&(p1->next[l+1]), s, e)) {  // insert in list l+1
+    if (!p1->next[l+1].cmpxch(s, e)) {  // insert in list l+1
-        (void) __sync_fetch_and_and(&(e->next[l+1]), (struct range *)0x1);    // failed, keep mark bit
+      e->next[l+1].ptr() = 0;    // failed, keep mark bit
      e->curlevel--;
      // cprintf("%d: crange_add_index: retry add level %d %u(%u)\n", mycpu()->id, l+1, e->key, e->key+e->size);
      //INCRETRY;
@@ -439,7 +433,7 @@ int crange::lock_range(u64 k, u64 sz, int l, range **er, range **pr, range **fr,
    if (l == 0) {
      acquire(e->lock);    // lock all ranges in the sequence
    }
-    e = WOMARK(e->next[l]);
+    e = e->next[l].ptr();
  }
  *sr = e;
  *er = e;
@@ -463,7 +457,7 @@ int crange::find_and_lock(u64 k, u64 sz, range **p0, range **f0, range **l0, ran
    p1 = *p0;   // remember last previous (p0) as the previous one level up (p1)
    *p0 = (l == this->nlevel-1) ? this->crange_head : p1;   // set current previous
    s1 = *s0;
-    for (e = WOMARK((*p0)->next[l]); e; *p0 = e, e = WOMARK(e->next[l])) {
+    for (e = (*p0)->next[l].ptr(); e; *p0 = e, e = e->next[l].ptr()) {
      assert(l < e->nlevel);
      int r = this->del_index(*p0, &e, l);
      if (r == -1) goto retry;   // deletion failed because some other core did it; try again
@@ -512,13 +506,13 @@ range* crange::search(u64 k, u64 sz, int mod)
  r = nullptr;
  p = this->crange_head;
  for (int l = this->nlevel-1; l >= 0; l--) {
-    for (e = WOMARK(p->next[l]); e; p = e, e = WOMARK(e->next[l])) {
+    for (e = p->next[l].ptr(); e; p = e, e = e->next[l].ptr()) {
      if (crange_debug)
 	cprintf("level %d: 0x%lx 0x%lx-%lx(%lu) 0x%lx-0x%lx(%lu)\n", l, (u64) p, p->key, p->key+p->size, p->size, e->key, e->key+e->size, e->size);
      // skip all marked ranges, but don't update p because
      // we don't want to descend on a marked range down.
-      while (e && MARKED(e->next[l])) {
+      while (e && e->next[l].mark()) {
-	e = WOMARK(e->next[l]);
+	e = e->next[l].ptr();
      }
      if (!e) break;
      if (mod && l < n && l > 0) {
@@ -565,7 +559,7 @@ void crange::del(u64 k, u64 sz)
  while (1) {
    // hook new list into bottom list; if del resulted in a new list, use that (repl), otherwise
    // set predecessor to successor.
-    if (__sync_bool_compare_and_swap(&(prev->next[0]), first, repl ? repl : succ)) {
+    if (prev->next[0].cmpxch(first, repl ? repl : succ)) {
      release(prev->lock);
      freen(first, last);   // put on delayed list before unlocking
      unlockn(first, last);
@@ -605,8 +599,8 @@ void crange::add(u64 k, u64 sz, void *v)
  repl = insert(repl, r);
  mark(first, succ);
  if (prev)
-    assert(!MARKED(prev->next[0]));
+    assert(!prev->next[0].mark());
-  if (__sync_bool_compare_and_swap(&(prev->next[0]), first ? first : succ, repl)) {
+  if (prev->next[0].cmpxch(first ? first : succ, repl)) {
    release(prev->lock);
    freen(first, last);   // put on delayed list before unlocking
    unlockn(first, last);
@@ -621,7 +615,7 @@ void crange::add(u64 k, u64 sz, void *v)
 int crange::foreach(int (*cb)(range *r, void *), void *st)
 {
  struct range *e;
-  for (e = WOMARK(this->crange_head->next[0]); e; e = WOMARK(e->next[0])) {
+  for (e = crange_head->next[0].ptr(); e; e = e->next[0].ptr()) {
    if (!cb(e, st))
      return 0;
  }

--- a/crange.hh
+++ b/crange.hh
@@ -5,6 +5,82 @@
 using std::atomic;
 struct crange;
+struct range;
+template<class T>
+class markptr_ptr;
+template<class T>
+class markptr_mark;
+template<class T>
+class markptr {
+ protected:
+  atomic<uptr> _p;
+ public:
+  markptr() : _p(0) {}
+  markptr(T* v) : _p((uptr) v) {}
+  markptr(const markptr<T> &v) : _p(v._p.load()) {}
+  void operator=(T* v) { _p = (uptr) v; }
+  void operator=(const markptr<T> &v) { _p = v._p.load(); }
+  bool operator!=(const markptr<T> &v) const { return _p != v._p; }
+  bool operator==(const markptr<T> &v) const { return _p == v._p; }
+  markptr_ptr<T>& ptr() {
+    return *(markptr_ptr<T>*) this;
+  }
+  markptr_mark<T>& mark() {
+    return *(markptr_mark<T>*) this;
+  }
+  // Convenience operator to avoid having to write out xx.ptr()->...
+  T* operator->() { return ptr(); }
+  bool cmpxch(markptr<T> expected, markptr<T> desired) {
+    uptr ee = expected._p.load();
+    return _p.compare_exchange_weak(ee, desired._p.load());
+  }
+};
+template<class T>
+class markptr_ptr : private markptr<T> {
+ public:
+  void operator=(T *ptr) {
+    uptr p0, p1;
+    do {
+      p0 = markptr<T>::_p.load();
+      p1 = (p0 & 1) | (uptr) ptr;
+    } while (!markptr<T>::_p.compare_exchange_weak(p0, p1));
+  }
+  T* load() const {
+    return (T*) (markptr<T>::_p.load() & ~1);
+  }
+  operator T*() const { return load(); }
+};
+template<class T>
+class markptr_mark : public markptr<T> {
+ public:
+  void operator=(bool mark) {
+    uptr p0, p1;
+    do {
+      p0 = markptr<T>::_p.load();
+      p1 = (p0 & ~1) | !!mark;
+    } while (!markptr<T>::_p.compare_exchange_weak(p0, p1));
+  }
+  bool load() const {
+    return markptr<T>::_p.load() & 1;
+  }
+  operator bool() const { return load(); }
+};
 struct range : public rcu_freed {
 public:
@@ -14,9 +90,9 @@ public:
  atomic<int> curlevel;  // the current levels it appears on
  int nlevel;            // the number of levels this range should appear
  crange *cr;            // the crange this range is part of
-  struct range** next;   // one next pointer per level
+  markptr<range>* next;  // one next pointer per level
-  struct spinlock *lock; // on separate cache line?
+  spinlock *lock;        // on separate cache line?
-  range(crange *cr, u64 k, u64 sz, void *v, struct range *n, int nlevel);
+  range(crange *cr, u64 k, u64 sz, void *v, markptr<range> n, int nlevel = 0);
  ~range();
  virtual void do_gc() {
    delete this;
@@ -24,7 +100,7 @@ public:
  void print(int l);
  void free_delayed();
  void dec_ref(void);
-  int lockif(range *e);
+  int lockif(markptr<range> e);
 } __mpalign__;
 struct crange {
@@ -41,7 +117,7 @@ public:
  void print(int);
  void check(struct range *absent);
  int del_index(range *p0, range **e, int l);
-  void add_index(int l, range *e, range *p1, range *s1);
+  void add_index(int l, range *e, range *p1, markptr<range> s1);
  int lock_range(u64 k, u64 sz, int l, range **er, range **pr, range **fr, range **lr, range **sr);
  int find_and_lock(u64 k, u64 sz, range **p0, range **f0, range **l0, range **s0);
 };