11
\$\begingroup\$

I'm trying to write a lock-free implementation for atomic shared pointer. Basically, there are two class templates shared_ptr and atomic_shared_ptr, each implemented in a header file of its own, for a total of two header files. The algorithm is inspired by the split reference counts idea used to implement a lock-free stack in C++ Concurrency in Action Sect. 7.2.4. This is my first time designing and writing lock-free algorithms, and I'm not confident. I want to have my code reviewed for possible flaws in the algorithm (quite likely ~>_<~) and room for improvement. Note that

  1. I don't intend to faithfully follow the interface specified by the standard/draft, because it's too cumbersome. Specifically, I didn't implement some methods and modified the signature of some others.
  2. Refining memory order requirements is left to future work.

Code

shared_ptr.hpp (view on GitHub for better syntax highlighting)

namespace shared_ptr_impl {

// cnt encodes both reference count and staged reference count.
// Staged reference count is used to support atomic_shared_ptr.
// *this can be deleted iff both reach zero.
// Specifically, cnt = (staged reference count) * 2^32 + (reference count).
// So long as reference count < 2^32, the two should not interfere with each other.
// Use unsigned type to wrap-around and avoid overflow.
// Encodes the two into a single 64-bit variable cnt, so that operations on the two
// can be made a single atomic operation on cnt.
// pdata is only modified at construction/destruction.
// This struct is therefore thread-safe.
template <typename T>
struct block {
  std::unique_ptr<T> pdata;
  std::atomic_uint64_t cnt;
};

} // namespace shared_ptr_impl

template <typename T>
class shared_ptr {
  template <typename U>
  friend class atomic_shared_ptr;

  using unique_ptr = std::unique_ptr<T>;
  using block = shared_ptr_impl::block<T>;

public:
  // copy control
  shared_ptr(const shared_ptr& p):
    pblock(p.pblock) {
    if (pblock) ++pblock->cnt;
  }

  shared_ptr(shared_ptr&& p) noexcept:
    pblock(std::exchange(p.pblock, nullptr)) {
    // pass
  }

  shared_ptr& operator=(shared_ptr p) noexcept {
    swap(*this, p);
    return *this;
  }

 ~shared_ptr() {
    if (pblock) {
      if (--pblock->cnt == 0) delete pblock;
    }
  }

  friend void swap(shared_ptr& a, shared_ptr& b) {
    std::swap(a.pblock, b.pblock);
  }

  // construct
  shared_ptr(T* p = nullptr):
    pblock(!p ? nullptr : new block{unique_ptr(p), {1}}) {
    // pass
  }

  // modifier
  void reset(T* p = nullptr) {
    auto expire(std::move(*this));
    pblock = !p ? nullptr : new block{unique_ptr(p), {1}};
  }

  // observer
  T* get() const {
    return !pblock ? nullptr : pblock->pdata.get();
  }

  T& operator*() const {
    assert(*this);
    return *pblock->pdata;
  }

  T* operator->() const {
    assert(*this);
    return get();
  }

  explicit operator bool() const {
    return pblock;
  }

  bool is_lock_free() const {
    assert(*this);
    return pblock->cnt.is_lock_free();
  }

private:
  shared_ptr(block* p):
    pblock(p) {
    // pass
  }

  block* pblock;
};

atomic_shared_ptr.hpp (view on GitHub)

namespace atomic_shared_ptr_impl {

using shared_ptr_impl::block;

// Works with block.
// Actual staged reference count = stagecnt / one_stagecnt (i.e., uint64_t(1) << 32).
template <typename T>
struct counted_ptr {
  std::uint64_t stagecnt;
  block<T>* pblock;
};

} // namespace atomic_shared_ptr_impl

template <typename T>
class atomic_shared_ptr {
  using shared_ptr_t = shared_ptr<T>;
  using counted_ptr = atomic_shared_ptr_impl::counted_ptr<T>;
  static constexpr auto one_stagecnt = std::uint64_t(1) << 32;

public:
  // copy control
  atomic_shared_ptr(const atomic_shared_ptr&) = delete;
  atomic_shared_ptr& operator=(const atomic_shared_ptr&) = delete;
 ~atomic_shared_ptr() {
    auto p = pblock.load();
    if (p.pblock) {
      p.pblock->cnt += p.stagecnt;
      shared_ptr_t(p.pblock);
    }
  }

  // construct
  atomic_shared_ptr():
    pblock{} {
    // pass
  }

  atomic_shared_ptr(shared_ptr_t p):
    pblock{0, std::exchange(p.pblock, nullptr)} {
    // pass
  }

  // modify
  void operator=(shared_ptr_t p) {
    counted_ptr newp{0, std::exchange(p.pblock, nullptr)};
    auto oldp = pblock.exchange(newp);
    if (oldp.pblock) {
      oldp.pblock->cnt += oldp.stagecnt;
      p.pblock = oldp.pblock;
    }
  }

  shared_ptr_t exchange(shared_ptr_t p) {
    counted_ptr newp{0, std::exchange(p.pblock, nullptr)};
    auto oldp = pblock.exchange(newp);
    if (oldp.pblock) oldp.pblock->cnt += oldp.stagecnt;
    return oldp.pblock;
  }

  bool compare_exchange_weak(shared_ptr_t& expect, const shared_ptr_t& desire) {
    auto oldp = copy_ptr();
    if (oldp.pblock != expect.pblock) {
      expect = oldp.pblock;
      return false;
    }
    counted_ptr newp{0, desire.pblock};
    if (pblock.compare_exchange_strong(oldp, newp)) {
      if (oldp.pblock) oldp.pblock->cnt += oldp.stagecnt - 2;
      if (desire) ++desire.pblock->cnt;
      return true;
    }
    if (expect) --expect.pblock->cnt;
    return false;
  }

  bool compare_exchange_weak(shared_ptr_t& expect, shared_ptr_t&& desire) {
    auto oldp = copy_ptr();
    if (oldp.pblock != expect.pblock) {
      expect = oldp.pblock;
      return false;
    }
    counted_ptr newp{0, desire.pblock};
    if (pblock.compare_exchange_strong(oldp, newp)) {
      if (oldp.pblock) oldp.pblock->cnt += oldp.stagecnt - 2;
      desire.pblock = nullptr;
      return true;
    }
    if (expect) --expect.pblock->cnt;
    return false;
  }

  bool compare_exchange_strong(shared_ptr_t& expect, const shared_ptr_t& desire) {
    auto p = expect.pblock;
    while (!compare_exchange_weak(expect, desire) && p == expect.pblock);
    return p == expect.pblock;
  }

  bool compare_exchange_strong(shared_ptr_t& expect, shared_ptr_t&& desire) {
    auto p = expect.pblock;
    while (!compare_exchange_weak(expect, std::move(desire)) && p == expect.pblock);
    return p == expect.pblock;
  }

  // observer
  bool is_lock_free() const {
    return pblock.is_lock_free();
  }

  operator shared_ptr_t() const {
    auto p = copy_ptr();
    return p.pblock;
  }

private:
  mutable std::atomic<counted_ptr> pblock;

  counted_ptr copy_ptr() const {
    counted_ptr p = pblock, pp;
    do {
      if (!p.pblock) return p;
      pp = p;
      pp.stagecnt += one_stagecnt;
    }
    while (!pblock.compare_exchange_weak(p, pp));
    pp.pblock->cnt += -one_stagecnt + 1;
    return pp;
  }
};
\$\endgroup\$
  • 1
    \$\begingroup\$ Very interesting. I think the two .h files should be combined; They are intimately related. Some demo code with a main() would help a lot, ideally with no assembly required. \$\endgroup\$ – Jive Dadson Feb 24 '18 at 1:20
  • 1
    \$\begingroup\$ @JiveDadson Agreed. i will consider adding some demo/test code later. \$\endgroup\$ – Lingxi Feb 24 '18 at 2:59
  • \$\begingroup\$ Hey there, this looks interesting but the github links are dead, can you post it somewhere else? Did you ever finish the demo code? \$\endgroup\$ – jrh Nov 6 '18 at 0:18
  • \$\begingroup\$ FYI, it looks like this is the OP's github repo. \$\endgroup\$ – jrh Nov 6 '18 at 18:46
1
\$\begingroup\$

Looks reasonable to me. But I strongly recommend splitting up some of the more complicated source lines to conform to the rule "one side effect, one line of code." For example:

if (--pblock->cnt == 0) delete pblock;

This really means

uint64_t new_cnt = --pblock->cnt;
if (new_cnt == 0) {
    delete pblock;
}

Or again:

void reset(T* p = nullptr) {
    auto expire(std::move(*this));
    pblock = !p ? nullptr : new block{unique_ptr(p), {1}};
}

This really means

void reset(T *p) {
    shared_ptr expire = std::move(*this);
    if (p != nullptr) {
        pblock = new block{unique_ptr(p), {1}};
    }
}
void reset() {
    shared_ptr expire = std::move(*this);
}

or possibly just

void reset(T *p) {
    swap(this, shared_ptr(p));
}
void reset() {
    reset(nullptr);
}

private:
  shared_ptr(block* p):
    pblock(p) {
    // pass
  }

Always, always, always mark your constructors explicit! This goes quadruple for implementation-detail private constructors.

In this case, because you failed to provide a shared_ptr(std::nullptr_t) constructor, you won't be able to call

void foo(shared_ptr<Widget>);
void test() {
    foo(nullptr);
}

because there are two equally good ways of converting nullptr to shared_ptr<Widget> — via the public T* constructor and via the private block* constructor — and so you get an ambiguity. Marking the private constructor explicit would have solved that particular problem. However, you'll still need the constructor from std::nullptr_t (a.k.a. decltype(nullptr)) in order to solve

void foo(shared_ptr<Widget>);
void test() {
    foo(shared_ptr<Widget>(nullptr));
}

Your indentation of destructors by n-1 characters is idiosyncratic and weird.

    foo();
   ~foo();  // Strange!

 ~atomic_shared_ptr() {
    auto p = pblock.load();
    if (p.pblock) {
      p.pblock->cnt += p.stagecnt;
      shared_ptr_t(p.pblock);
    }
  }

The line

      shared_ptr_t(p.pblock);

may be correct, but it certainly looks strange. I would also question the use of auto on your first line, given how important it is in this code that we know which of our variables are atomic and which aren't. So I would write this as

  ~atomic_shared_ptr() {
      counted_ptr p = pblock.load();
      if (p.pblock != nullptr) {
          std::atomic_uint64_t& cnt = p.pblock->cnt;
          cnt += p.stagecnt;
          std::uint64_t old_cnt = cnt--;
          if (old_cnt == 1) {
              delete p.pblock;
          }
      }
  }

And then I would wonder whether it was possible to coalesce those two additions into

      std::uint64_t new_cnt = (cnt += p.stagecnt - 1);
      if (new_cnt == 0) {

operator shared_ptr_t() const {
    auto p = copy_ptr();
    return p.pblock;
}

Again with the implicit conversion. p.pblock isn't a shared_ptr<T>, so it's strange to be returning it from a function that claims to return shared_ptr<T>. Eliminating that implicit conversion from block* to shared_ptr will force you to fix this too.


counted_ptr p = pblock, pp;

I strongly recommend writing one declaration per line.

counted_ptr p = pblock;
counted_ptr pp;

It seems like the bulk of the "interesting" stuff (and thus the highest likelihood of bugs) is in the compare_exchange_weak functions; and they depend on the copy_ptr function, which is non-standard (so cppreference won't help the reader understand its purpose) and undocumented. I strongly recommend adding some commentary to copy_ptr that explains what its preconditions and postconditions are supposed to be.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.