一、为什么要用DoublyBufferedData

LoadBalancer是一个读远多于写的数据结构：大部分时候，所有线程从一个不变的server列表中选取一台server。如果server列表真是“不变的”，那么选取server的过程就不用加锁，我们可以写更复杂的分流算法。一个方法是用读写锁，但当读临界区不是特别大时（毫秒级），读写锁并不比mutex快，而实用的分流算法不可能到毫秒级，否则开销也太大了。另一个方法是双缓冲，很多检索端用类似的方法实现无锁的查找过程，它大概这么工作：

1、数据分前台和后台

2、检索线程只读前台，不用加锁

3、只有一个写线程：修改后台数据，切换前后台，睡眠一段时间，以确保老前台（新后台）不再被检索线程访问

这个方法的问题在于它假定睡眠一段时间后就能避免和前台读线程发生竞争，这个时间一般是若干秒。由于多次写之间有间隔，这儿的写往往是批量写入，睡眠时正好用于积累数据增量。

二、DoublyBufferedData源码

在DoublyBufferedData内部包含Wrapper类（定义在外部）、ScopedPtr类、Read方法、Modify方法、AddWrapper方法、RemoveWrapper方法、存储前后台数据的数组_data、前台数据的索引_index、访问Wrapper的key、所有数据实例_wrappers、批量访问Wrappers的锁、批量写入锁_modify_mutex

1、Wrapper类

Wrapper类继承自DoublyBufferedDataWrapperBase类，后者只包含一个用户定义的数据类型以及获取该数据的方法。

在 Wrapper类中主要包含两个私有变量，即

template <typename T, typename TLS>
class DoublyBufferedData<T, TLS>::Wrapper
    : public DoublyBufferedDataWrapperBase<T, TLS> {
friend class DoublyBufferedData;
public:
    explicit Wrapper(DoublyBufferedData* c) : _control(c) {
        pthread_mutex_init(&_mutex, NULL);
    }
    
    ~Wrapper() {
        if (_control != NULL) {
            _control->RemoveWrapper(this);
        }
        pthread_mutex_destroy(&_mutex);
    }

    // _mutex will be locked by the calling pthread and DoublyBufferedData.
    // Most of the time, no modifications are done, so the mutex is
    // uncontended and fast.
    inline void BeginRead() {
        pthread_mutex_lock(&_mutex);
    }

    inline void EndRead() {
        pthread_mutex_unlock(&_mutex);
    }

    inline void WaitReadDone() {
        BAIDU_SCOPED_LOCK(_mutex);
    }
    
private:
    DoublyBufferedData* _control;
    pthread_mutex_t _mutex;
};

DoublyBuffedData是Wrapper类中的友元类，后者又是前者的成员类型。

Wrapper类的构造方法主要是申请一把锁，析构函数主要负责释放掉Wrapper以及释放锁。因为大部分情况下，对于数据的修改不会很频繁，所以该锁大部分情况下不会阻塞而且速度很快

除此之外、Wrapper类中主要包含三个方法，BeginRead()方法、EndRead()方法、WaitReadDone()方法。

2、AddWrapper方法和RemoveWrapper方法

AddWrapper方法主要在线程初始化Wrapper的时候调用，主要负责向_wrappers向量中增加新的Wrapper

// Called when thread initializes thread-local wrapper.
template <typename T, typename TLS>
typename DoublyBufferedData<T, TLS>::Wrapper*
DoublyBufferedData<T, TLS>::AddWrapper() {
    std::unique_ptr<Wrapper> w(new (std::nothrow) Wrapper(this));
    if (NULL == w) {
        return NULL;
    }
    try {
        BAIDU_SCOPED_LOCK(_wrappers_mutex);
        _wrappers.push_back(w.get());
    } catch (std::exception& e) {
        return NULL;
    }
    return w.release();
}

RemoveWrapper方法主要在线程退出的时候调用，负责删除对应的Wrapper

// Called when thread quits.
template <typename T, typename TLS>
void DoublyBufferedData<T, TLS>::RemoveWrapper(
    typename DoublyBufferedData<T, TLS>::Wrapper* w) {
    if (NULL == w) {
        return;
    }
    BAIDU_SCOPED_LOCK(_wrappers_mutex);
    for (size_t i = ; i < _wrappers.size(); ++i) {
        if (_wrappers[i] == w) {
            _wrappers[i] = _wrappers.back();
            _wrappers.pop_back();
            return;
        }
    }
}

3、Read方法

在Read方法中，首先判断是否创建了访问Wrapper的key，然后通过该key获取对应的Wrapper（如果查不到，则调用AddWrapper初始化一个Wrapper并赋值）

template <typename T, typename TLS>
int DoublyBufferedData<T, TLS>::Read(
    typename DoublyBufferedData<T, TLS>::ScopedPtr* ptr) {
    if (BAIDU_UNLIKELY(!_created_key)) {
        return -1;
    }
    Wrapper* w = static_cast<Wrapper*>(pthread_getspecific(_wrapper_key));
    if (BAIDU_LIKELY(w != NULL)) {
        w->BeginRead();
        ptr->_data = UnsafeRead();
        ptr->_w = w;
        return ;
    }
    w = AddWrapper();
    if (BAIDU_LIKELY(w != NULL)) {
        const int rc = pthread_setspecific(_wrapper_key, w);
        if (rc == ) {
            w->BeginRead();
            ptr->_data = UnsafeRead();
            ptr->_w = w;
            return ;
        }
    }
    return -1;
}

4、Modify方法

在调用Modify方法的时候，会单独创建一个单独的线程用来修改后台的数据（通常修改操作不会特别频繁，所以只用一个单独的线程即可），修改之后会调换前后台，等待老前台数据全部读取结束之后，修改新后台数据（老前台数据）

template <typename T, typename TLS>
template <typename Fn>
size_t DoublyBufferedData<T, TLS>::Modify(Fn& fn) {
    // _modify_mutex sequences modifications. Using a separate mutex rather
    // than _wrappers_mutex is to avoid blocking threads calling
    // AddWrapper() or RemoveWrapper() too long. Most of the time, modifications
    // are done by one thread, contention should be negligible.
    BAIDU_SCOPED_LOCK(_modify_mutex);
    int bg_index = !_index.load(butil::memory_order_relaxed);
    // background instance is not accessed by other threads, being safe to
    // modify.
    const size_t ret = fn(_data[bg_index]);
    if (!ret) {
        return ;
    }

    // Publish, flip background and foreground.
    // The release fence matches with the acquire fence in UnsafeRead() to
    // make readers which just begin to read the new foreground instance see
    // all changes made in fn.
    _index.store(bg_index, butil::memory_order_release);
    bg_index = !bg_index;
    
    // Wait until all threads finishes current reading. When they begin next
    // read, they should see updated _index.
    {
        BAIDU_SCOPED_LOCK(_wrappers_mutex);
        for (size_t i = ; i < _wrappers.size(); ++i) {
            _wrappers[i]->WaitReadDone();
        }
    }

    const size_t ret2 = fn(_data[bg_index]);
    CHECK_EQ(ret2, ret) << "index=" << _index.load(butil::memory_order_relaxed);
    return ret2;
}

该篇文章还会持续更新，欢迎感兴趣的小伙伴关注专栏：

独立团丶zhuanlan.zhihu.com

欢迎感兴趣的小伙伴关注公众号：

ITPub博客

BRPC详解（四）——负载均衡之DoublyBufferedData

一、为什么要用DoublyBufferedData

二、DoublyBufferedData源码

最新文章