实现一个strong_rc_ptr(非线程安全版本的std::shared_ptr)

owent

发布于 2024-10-09 06:32:50

18100

代码可运行

文章被收录于专栏：owentowent

运行总次数：0

代码可运行

背景

我们的新项目有个比较复杂的全区全服交易行系统，其中搜索和推荐是高实时性全区服多维度排序的，并且要支持比较复杂的标签交集查询和属性范围查询的自由组合。当有订单发生变化时，它不仅仅会影响全服状态下搜索和推荐条件的结果变化，也会同时影响商品维度的聚合，交易行层面的数据聚合。

为了降低搜索开销和提供弹性伸缩的能力，我们在多个不同类型的服务上设计了基于视图的动态索引支持，并且针对常用的一些搜索推荐模型，实现了大量的静态索引。

这也同时带来的影响是每当订单有变化时它都有可能会去刷新大量的索引。我们早期第一版直接用 std::shared_ptr 来维护订单信息。每次变更索引时都是重新入删除和插入一个 std::shared_ptr 。 std::shared_ptr 底层的实现是使用 std::atomic 来维护了引用计数。每次变更操作都会导致 std::atomic 增减，同时带来 CPU Cache Line 失效。

在大多数场景下，这也没太大问题，因为大多数场景下参数传递 std::shared_ptr 也好，内存存放 std::shared_ptr 也好，通常上下文会存在更多逻辑操作，导致这个CPU Cache Line 失效后重新prefetch的开销占比很小。但是在我们这个场景下情况又有所不同。我们采用了BTree来管理有序索引，这样比常规的红黑树来说内存结构会更紧凑，缓存命中率更高。然而在连续大量索引发生变化时，一直要发生Cache失效->重新prefetch，Cache失效->重新prefetch的过程。这反而带来了额外的浪费。我们在后来用valgrind分析的过程中也确实验证导这部分的Cache Miss率明显高于其他操作。于是实现一个非线程安全版本的 shared_ptr 就被提上了日程。思路其实和 Rust的 std::rc::Rc 很像。

其实GCC的 STL本身自带费线程安全的内部版本的，可以通过使用 template<class T> using strong_rc_ptr<T> = std::__shared_ptr<T, std::_S_single>; 然后使用 strong_rc_ptr<T> 就是单线程模式。然而我们这里考虑到跨编译器跨平台，所以自己实现了一个。

实现

基础功能

引用计数型的智能指针的基本原理比较简单。就是有一个存储区去存放引用计数。当计数清零后释放。

为了加快解引用的性能，原始指针并没有放在引用计数的存储区里，而是直接放在智能指针对象上。这样大多数场景访问指针内容的时候不需要多一次跳转去查询实际地址。

接下来更多的代码其实是在适配和优化各种使用场景。首先是针对各类构造场景，我的实现分成了5种。（ std::shared_ptr 和 boost::shared_ptr 分别按各自的实现有一些实现合并和分离，和我这边实现稍有差异，但主体Feature相同）

默认构造函数：接收指针传入，通过 new/delete 管理计数对象。
默认inplace构造：通过 new/delete 管理计数对象和实际对象的内存，通过placement new构造对象。(碎片更少， make_shared/make_strong_rc 的构造方式)
默认带自定义Deletor的构造：比上面的构造多存储一个Deletor对象，通过 new/delete 管理计数对象。
指定Allocator的inplace构造：通过自定义Allocator管理计数对象和实际对象的内存，通过placement new构造对象。（自定义分配器， allocate_shared/allocate_strong_rc 的构造方式）boost::shared_ptr 的实现有问题，某些地方显示使用 new/delete 操作符了，导致对自定义Allocator没有完整的支持。
指定Allocator的带自定义Deletor的构造：比上面的构造多存储一个Deletor对象，自定义Allocator管理计数对象。

有些实现里带Deletor和带Allocator的实现是合并的，我这里分离出来了。是因为没有Deletor的时候可以少一哥对象的维护操作，理论上会更利于编译优化。

另外 boost::shared_ptr 的某些分支实现显示使用 new/delete 操作符来分配管理对象的内存结构了，没有使用 allocator_traits<Alloc>::rebind_alloc<T> 这种形式。实际上也是不对的。所以这里我采用和 std::shared_ptr 一样的方式保持对自定义Allocator的完全兼容。具体详情也可以参考我之前写的《手夯一个STL allocator和对象内存分析组件》的Allocator rebind章节。

`enable_shared_from_this` 的两种实现

接下来是 enable_shared_from_this 的实现。要实现继承 enable_shared_from_this 的对象自动带 share_from_this() 接口，首先基类需要记录一个 weak_ptr/weak_rc_ptr 。然后构造 shared_ptr 的时候检测目标指针可否转换成 enable_shared_from_this 指针。这个检测目前新版本的编译器和STL实现大多数采用 std::void_t<T> 转换的detection idiom来实现。比如GCC实现:

private:
  template<typename _Tp1>
  void _M_weak_assign(_Tp1* __p, const __shared_count<_Lp>& __n) const noexcept {
    _M_weak_this._M_assign(__p, __n);
  }

  friend const __enable_shared_from_this*  __enable_shared_from_this_base(
    const __shared_count<_Lp>&, const __enable_shared_from_this* __p) {
    return __p;
  }

  // ...
  template<typename _Yp>
  using __esft_base_t = decltype(__enable_shared_from_this_base(
      std::declval<const __shared_count<_Lp>&>(),
      std::declval<_Yp*>()));

  // Detect an accessible and unambiguous enable_shared_from_this base.
  template<typename _Yp, typename = void>
  struct __has_esft_base : false_type { };

  template<typename _Yp>
  struct __has_esft_base<_Yp, __void_t<__esft_base_t<_Yp>>> : __not_<is_array<_Tp>> { }; // No enable shared_from_this for arrays

  template<typename _Yp, typename _Yp2 = typename remove_cv<_Yp>::type>
  typename enable_if<__has_esft_base<_Yp2>::value>::type
  _M_enable_shared_from_this_with(_Yp* __p) noexcept {
    if (auto __base = __enable_shared_from_this_base(_M_refcount, __p))
      __base->_M_weak_assign(const_cast<_Yp2*>(__p), _M_refcount);
  }

  template<typename _Yp, typename _Yp2 = typename remove_cv<_Yp>::type>
  typename enable_if<!__has_esft_base<_Yp2>::value>::type
  _M_enable_shared_from_this_with(_Yp*) noexcept { }

MSVC的STL也是这个实现，但是这个idiom在C++17之后一些行为才被标准化，老的编译器可能会不支持。所以我这里还是采用了更传统的实现方法，即采用模板匹配比动态参数函数更优先匹配的机制。

template <class T1, class T2, class T3>
inline void __enable_shared_from_this_with(const __strong_rc_counter<T1>* __n,
                                                                          const T2* __py,
                                                                          const enable_shared_rc_from_this<T3>* __p) {
  if (nullptr != __p) {
    __p->__internal_weak_assign(const_cast<T2*>(__py), *__n);
  }
}

template <class T1, class T2, class T3, size_t T3SIZE>
inline void __enable_shared_from_this_with(
    const __strong_rc_counter<T1>* __n, const T2* __py, const enable_shared_rc_from_this<T3[T3SIZE]>* __p) {
  if (nullptr != __p) {
    for (auto& p : *__p) {
      p->__internal_weak_assign(const_cast<T2*>(__py), *__n);
    }
  }
}

#ifdef _MANAGED

// Avoid C4793, ... causes native code generation

struct __sp_any_pointer {
  template <class T>
  __sp_any_pointer(T*) {}  // NOLINT: runtime/explicit
};

inline void __enable_shared_from_this_with(__sp_any_pointer, __sp_any_pointer,
                                                                          __sp_any_pointer) {}

#else  // _MANAGED

inline void __enable_shared_from_this_with(...) {}

#endif  // _MANAGED

const类型比较操作符

在实现操作符重载的时候有个小tips需要注意。比如如果比较 shared_ptr<T> 和 shared_ptr<const T> 的时候。如果实现的不好，容易在某些SFINAE流程里推断成 T* ，然后由于 const T* 不能转成 T* 导致推断失败。这个需要稍微注意一下。

`std::shared_ptr` 和 `boost::shared_ptr` 的差异

在写单元测试的时候，我发现 std::shared_ptr 和 boost::shared_ptr 的实现上还有一些行为上的差异。虽然说这部分可能是未定义行为，但是我觉得 std::shared_ptr 的行为大部分场景下更符合直觉。

空指针引用计数

// boost::shared_ptr
class shared_count
{
private:
  sp_counted_base * pi_;

public:
  BOOST_CONSTEXPR shared_count() BOOST_SP_NOEXCEPT: pi_(0)
#if defined(BOOST_SP_ENABLE_DEBUG_HOOKS)
      , id_(shared_count_id)
#endif
  {
  }

  long use_count() const BOOST_SP_NOEXCEPT {
    return pi_ != 0? pi_->use_count(): 0;
  }
};

// GCC: std::shared_ptr
template<_Lock_policy _Lp = __default_lock_policy>
class _Sp_counted_base : public _Mutex_base<_Lp> {
public:
  _Sp_counted_base() noexcept : _M_use_count(1), _M_weak_count(1) { }

  long _M_get_use_count() const noexcept {
    // No memory barrier is used here so there is no synchronization
    // with other threads.
    return __atomic_load_n(&_M_use_count, __ATOMIC_RELAXED);
  }
};

// 单元测试
util::memory::strong_rc_ptr<int> pi;
// boost::shared_ptr 行为（strong_rc_ptr采用此行为）
CASE_EXPECT_TRUE(pi.use_count() == 0);
// std::shared_ptr 行为
CASE_EXPECT_TRUE(pi.use_count() == 1);

pi.reset(static_cast<int *>(nullptr));
CASE_EXPECT_TRUE(pi.use_count() == 1);

这里空指针 boost::shared_ptr 引用计数为0， std::shared_ptr 引用计数为1。我觉得0更符合直觉就按0处理了。

操作符的怪异行为

/// GCC: libstdc++
template<typename _Tp1, typename _Tp2, _Lock_policy _Lp>
inline bool operator==(const __shared_ptr<_Tp1, _Lp>& __a, const __shared_ptr<_Tp2, _Lp>& __b) noexcept {
  return __a.get() == __b.get();
}
#ifdef __cpp_lib_three_way_comparison
template<typename _Tp, typename _Up, _Lock_policy _Lp>
inline strong_ordering operator<=>(const __shared_ptr<_Tp, _Lp>& __a, const __shared_ptr<_Up, _Lp>& __b) noexcept {
  return compare_three_way()(__a.get(), __b.get());
}
// 其他相似的重载不再展示 ...
#else
template<typename _Tp, typename _Up, _Lock_policy _Lp>
inline bool operator<(const __shared_ptr<_Tp, _Lp>& __a, const __shared_ptr<_Up, _Lp>& __b) noexcept {
  using _Tp_elt = typename __shared_ptr<_Tp, _Lp>::element_type;
  using _Up_elt = typename __shared_ptr<_Up, _Lp>::element_type;
  using _Vp = typename common_type<_Tp_elt*, _Up_elt*>::type;
  return less<_Vp>()(__a.get(), __b.get());
}
// 其他相似的重载不再展示 ...
#endif

// boost::shared_ptr
template<class T, class U>
inline bool operator==(shared_ptr<T> const & a, shared_ptr<U> const & b) BOOST_SP_NOEXCEPT {
  return a.get() == b.get();
}

template<class T, class U>
inline bool operator<(shared_ptr<T> const & a, shared_ptr<U> const & b) BOOST_SP_NOEXCEPT {
    return a.owner_before(b);
}

// 单元测试
{
  util::memory::strong_rc_ptr<int> p1;
  util::memory::strong_rc_ptr<int> p2;
  p2.reset(nullptr);

  CASE_EXPECT_TRUE(p1 == p2);

  // std::shared_ptr 行为（strong_rc_ptr采用此行为）
  CASE_EXPECT_FALSE((p1 < p2 || p2 < p1));
  // boost::shared_ptr 行为
  CASE_EXPECT_TRUE((p1 < p2 || p2 < p1));
}

这里可以看到，当 p1 == p2 的时候直观上显然 p1 < p2 || p2 < p1 应该是 false 的。但是因为boost的实现 operator== 用了原始指针，但是 operator< 缺用的管理数据块的指针。两个不一致。这里 strong_rc_ptr 保持和 std::shared_ptr 的行为一致。

继承和父子转换和比较操作符

struct X {
  int dummy;
};

struct Y {
  int dummy2;
};

struct Z : public X, public virtual Y {};

// 单元测试
util::memory::strong_rc_ptr<Z> pz(new Z);
util::memory::strong_rc_ptr<X> px(pz);

CASE_EXPECT_TRUE(px.get() == pz.get());
CASE_EXPECT_TRUE(px == pz);

util::memory::strong_rc_ptr<Y> py(pz);

CASE_EXPECT_TRUE(py.get() == pz.get());
CASE_EXPECT_TRUE(py == pz);
CASE_EXPECT_FALSE(py < pz || pz < py);

// strong_rc_ptr 行为， std::shared_ptr 不允许编译通过
CASE_EXPECT_TRUE(px < py || py < px);
// boost::shared_ptr 行为，和下面操作符实现相关
CASE_EXPECT_FALSE(px < py || py < px);

util::memory::strong_rc_ptr<void> pvx(px);
util::memory::strong_rc_ptr<void> pvy(py);
util::memory::strong_rc_ptr<void> pvz(pz);

CASE_EXPECT_TRUE(pvx.get() != pvy.get());
CASE_EXPECT_TRUE(pvx != pvy);

// std::shared_ptr 行为（strong_rc_ptr采用此行为）
CASE_EXPECT_TRUE(pvx < pvy || pvy < pvx);
CASE_EXPECT_TRUE(pvy < pvz || pvz < pvy);

// boost::shared_ptr 行为，和下面操作符实现相关
CASE_EXPECT_FALSE(pvx < pvy || pvy < pvx);
CASE_EXPECT_FALSE(pvy < pvz || pvz < pvy);

对于父子类转换后再类型擦除后的比较操作。我也是觉得 std::shared_ptr 的行为更符合直觉，所以按 std::shared_ptr 的行为为准。

单元测试

单元测试我直接就扒了 boost.shared_ptr 的了。部分和 std::shared_ptr 差异的部分按 std::shared_ptr 的行为做了调整。这样覆盖场景应该是比较完善了。

周边组件迁移

一键切换组件

因为我这个 strong_rc_ptr 实现了 shared_ptr 的所有接口，所以替换起来非常简单。也提供了一套 traits 接口用于一键切换。这样也方便比较性能差异。代码大致上这样:

LIBATFRAME_UTILS_NAMESPACE_BEGIN
namespace memory {
enum class compat_strong_ptr_mode : int8_t {
  kStrongRc = 0,
  kStl = 1,
};

template <compat_strong_ptr_mode>
struct LIBATFRAME_UTILS_API_HEAD_ONLY compat_strong_ptr_function_trait;

template <>
struct LIBATFRAME_UTILS_API_HEAD_ONLY compat_strong_ptr_function_trait<compat_strong_ptr_mode::kStrongRc> {
  template <class Y>
  using shared_ptr = memory::strong_rc_ptr<Y>;

  template <class Y>
  using weak_ptr = memory::weak_rc_ptr<Y>;

  template <class Y>
  using enable_shared_from_this = memory::enable_shared_rc_from_this<Y>;

  template <class Y, class... ArgsT>
  static inline memory::strong_rc_ptr<Y> make_shared(ArgsT&&... args) {
    return memory::make_strong_rc<Y>(std::forward<ArgsT>(args)...);
  }

  template <class Y, class Alloc, class... TArgs>
  static inline memory::strong_rc_ptr<Y> allocate_shared(const Alloc& alloc, TArgs&&... args) {
    return memory::allocate_strong_rc<Y>(alloc, std::forward<TArgs>(args)...);
  }

  template <class Y, class F>
  static inline memory::strong_rc_ptr<Y> static_pointer_cast(F&& f) {
    return memory::static_pointer_cast<Y>(std::forward<F>(f));
  }

  template <class Y, class F>
  static inline memory::strong_rc_ptr<Y> const_pointer_cast(F&& f) {
    return memory::const_pointer_cast<Y>(std::forward<F>(f));
  }

#if defined(LIBATFRAME_UTILS_ENABLE_RTTI) && LIBATFRAME_UTILS_ENABLE_RTTI
  template <class Y, class F>
  static inline memory::strong_rc_ptr<Y> dynamic_pointer_cast(F&& f) {
    return memory::dynamic_pointer_cast<Y>(std::forward<F>(f));
  }
#endif
};

template <>
struct LIBATFRAME_UTILS_API_HEAD_ONLY compat_strong_ptr_function_trait<compat_strong_ptr_mode::kStl> {
  template <class Y>
  using shared_ptr = std::shared_ptr<Y>;

  template <class Y>
  using weak_ptr = std::weak_ptr<Y>;

  template <class Y>
  using enable_shared_from_this = std::enable_shared_from_this<Y>;

  template <class Y, class... ArgsT>
  static inline std::shared_ptr<Y> make_shared(ArgsT&&... args) {
    return std::make_shared<Y>(std::forward<ArgsT>(args)...);
  }

  template <class Y, class Alloc, class... TArgs>
  static inline std::shared_ptr<Y> allocate_shared(const Alloc& alloc, TArgs&&... args) {
    return std::allocate_shared<Y>(alloc, std::forward<TArgs>(args)...);
  }

  template <class Y, class F>
  static inline std::shared_ptr<Y> static_pointer_cast(F&& f) {
    return std::static_pointer_cast<Y>(std::forward<F>(f));
  }

  template <class Y, class F>
  static inline std::shared_ptr<Y> const_pointer_cast(F&& f) {
    return std::const_pointer_cast<Y>(std::forward<F>(f));
  }

#if defined(LIBATFRAME_UTILS_ENABLE_RTTI) && LIBATFRAME_UTILS_ENABLE_RTTI
  template <class Y, class F>
  static inline std::shared_ptr<Y> dynamic_pointer_cast(F&& f) {
    return std::dynamic_pointer_cast<Y>(std::forward<F>(f));
  }
#endif
};

template <class T, compat_strong_ptr_mode PtrMode>
struct LIBATFRAME_UTILS_API_HEAD_ONLY compat_strong_ptr_type_trait {
  using shared_ptr = typename compat_strong_ptr_function_trait<PtrMode>::template shared_ptr<T>;
  using weak_ptr = typename compat_strong_ptr_function_trait<PtrMode>::template weak_ptr<T>;
};

}  // namespace memory
LIBATFRAME_UTILS_NAMESPACE_END

算法和容器基础组件适配

有了上面的一键切换traits，其他的组件切换也就比较简单了。在 atframe_utils 内的组件接入一键切换的有时间轮定时器实现 jeffies_timer ， LRU算法容器 lru_map ，分布式系统和分布式事务的 WAL模块 。

还有一部分是我们工程里有很多读取策划通过配置Excel转出的数据，这部分访问量也很大。所以我对这个生成读表代码的模板系统 xres-code-generator 也做了改造支持。

效果

其实在完成之前，其实我们也不确定到底能带来多大的提升。简单的循环benchmark其实也没啥意义，因为和实际使用场景的访问情况相差太远了。

我们在完成之后对我们实际项目路14-16个静态索引的交易行上下架请求和搜索的场景做了对比（不包含Excel读表改造），大概比 std::shared_ptr 提升了10%-16%的综合性能，这里面其实附带了其他的一些视图和索引的比较操作和其他RPC的操作。所以单单这个智能指针的部分提升应该是更大的。