无锁动态扩容数组实现问题：高并发下崩溃与堆损坏求助

我正依据Stroustrup的无锁向量论文实现无锁数组，目标是构建线程安全的动态扩容数组。该实现在低负载下运行正常，但两个线程同时插入2e5个元素时，出现以下错误：

• Process finished with exit code 139 (interrupted by signal 11: SIGSEGV).，调试发现descriptor莫名变为null。
• 堆损坏错误：

ds(86913,0x16dde7000) malloc: Heap corruption detected, free list is damaged at 0x60000192c000
*** Incorrect guard value: 4329210352
ds(86913,0x16de73000) malloc: *** error for object 0x60000192c000: pointer being freed was not allocated
ds(86913,0x16de73000) malloc: *** set a breakpoint in malloc_error_break to debug

实现代码 #

LockFreeArray.h #

#ifndef DS_LOCKFREEARRAY_H
#define DS_LOCKFREEARRAY_H

#include <algorithm>
#include <cstddef>

namespace fast_ds {

    namespace internal {
        template<class T>
        class write_descriptor {
        public:
            T old_value_;
            T new_value_;
            size_t location_;
            bool completed_;

            write_descriptor() : location_(0), completed_(true) {}

            explicit write_descriptor(T oldValue, T newValue, size_t location, bool completed) : old_value_(oldValue),
                                                                                                 new_value_(newValue),
                                                                                                 location_(location),
                                                                                                 completed_(completed) {}
        };

        template<class T>
        class v_descriptor {
        public:
            size_t size_ = 0;
            write_descriptor<T> writeDescriptor_;

            v_descriptor() = default;

            explicit v_descriptor(size_t size, write_descriptor<T> w_descriptor) : size_(size),
                                                                                   writeDescriptor_(w_descriptor) {}
        };
    }

    template<class T>
    class LockFreeArray {
    private:
        static constexpr size_t kNumberOfBuckets = 32;
        static constexpr size_t kFirstBucketCapacity = 2;

        std::atomic<std::atomic<T> *> *data_;
        std::shared_ptr<internal::v_descriptor<T>> descriptor_ = std::make_shared<internal::v_descriptor<T>>();

        void CompleteWrite(internal::write_descriptor<T> write_descriptor) {
            if (!write_descriptor.completed_) {
                std::atomic<T> *memoryLocation = At(write_descriptor.location_);
                memoryLocation->compare_exchange_weak(write_descriptor.old_value_,
                                                      write_descriptor.new_value_);

                write_descriptor.completed_ = true;
            }
        }

        inline int HighestBitSet(unsigned int x) const {
            return 31 - __builtin_clz(x);
        }

        void AllocBucket(int bucket) {
            int bucketSize = 1 << (bucket + HighestBitSet(kFirstBucketCapacity));
            std::atomic<T> *currentBucket = data_[bucket].load();
            auto newBucket = new std::atomic<T>[bucketSize];

            if (!data_[bucket].compare_exchange_weak(currentBucket, newBucket)) {
                delete[] newBucket;
            }
        }

    public:
        explicit LockFreeArray() {
            data_ = new std::atomic<std::atomic<T> *>[kNumberOfBuckets];
            data_[0].store(new std::atomic<T>[kFirstBucketCapacity]);
        }

        LockFreeArray(const LockFreeArray &other) = delete;

        size_t Size() noexcept {
            auto current_descriptor = descriptor_;
            auto current_size = current_descriptor->size_;

            if (!descriptor_->writeDescriptor_.completed_) {
                current_size--;
            }

            return current_size;
        }

        std::atomic<T> *At(size_t index) const {
            auto pos = index + kFirstBucketCapacity;
            auto hibit = HighestBitSet(pos);
            auto index_in_bucket = (pos ^ (1 << hibit));
            return &data_[hibit - HighestBitSet(kFirstBucketCapacity)][index_in_bucket];
        }

        void PushBack(const T &value) {
            internal::v_descriptor<T> *current_descriptor;
            std::shared_ptr<internal::v_descriptor<T>> new_descriptor;
            std::shared_ptr<internal::v_descriptor<T>> current_descriptor_ptr = descriptor_;

            do {
                current_descriptor = descriptor_.get();
                if (current_descriptor == nullptr) {
                    continue;
                }
                current_descriptor_ptr = descriptor_;
                int current_size = current_descriptor->size_;
                internal::write_descriptor<T> wDescriptor = current_descriptor->writeDescriptor_;
                CompleteWrite(wDescriptor);

                if (int bucket = HighestBitSet(current_size + kFirstBucketCapacity) -
                                 HighestBitSet(kFirstBucketCapacity); data_[bucket].load() == 0) {
                    AllocBucket(bucket);
                }

                auto oldValue = At(current_size)->load();
                auto write_op = internal::write_descriptor<T>(oldValue, value,
                                                              current_size,
                                                              false);

                new_descriptor = std::make_shared<internal::v_descriptor<T>>(
                        internal::v_descriptor(current_size + 1, write_op));
            } while (!std::atomic_compare_exchange_strong(&descriptor_, &current_descriptor_ptr, new_descriptor));
            CompleteWrite(new_descriptor->writeDescriptor_);
        }
    };
}

#endif

主文件代码 #

#include <iostream>
#include <thread>
#include "include/vectors/LockFreeArray.h"

using namespace fast_ds;
using namespace std;

const int N = 100000;

auto v = LockFreeArray<int>();

void f1() {
    for(int i = 1; i <= N; ++i) {
        v.PushBack(i);
    }
}

void f2() {
    for(int i = 1 + N; i <= 2 * N; ++i) {
        v.PushBack(i);
    }
}

int main() {
    std::thread thread1(f1);
    std::thread thread2(f2);

    thread1.join();
    thread2.join();

    cout << v.Size() << endl;
}

补充说明 #

• 运行环境：M1芯片，Apple Clang 17
• 尝试过std::experimental::atomic_shared_ptr<T>和Clang20的std::atomic<std::shared_ptr<T>>，但当前编译器及GNU均未实现
• 为追求性能使用了裸指针，也可考虑切换为智能指针

1. 修复descriptor_的原子性问题 #

当前descriptor_是普通std::shared_ptr，直接用std::atomic_compare_exchange_strong操作属于未定义行为——std::shared_ptr的内部引用计数并非原子操作（除非使用std::atomic<std::shared_ptr>，但当前编译器不支持）。这是descriptor变为null的核心原因：多线程并发修改时，引用计数被破坏，导致v_descriptor提前析构。

替代方案：用原子指针管理v_descriptor #

// 替换原成员变量
std::atomic<internal::v_descriptor<T>*> descriptor_ = new internal::v_descriptor<T>();

// PushBack中的循环修改为：
internal::v_descriptor<T>* current_descriptor;
internal::v_descriptor<T>* new_descriptor;
do {
    current_descriptor = descriptor_.load(std::memory_order_acquire);
    if (!current_descriptor) continue;
    
    // 原有逻辑：完成未完成的写操作、检查并分配桶
    internal::write_descriptor<T> wDescriptor = current_descriptor->writeDescriptor_;
    CompleteWrite(wDescriptor);
    
    int current_size = current_descriptor->size_;
    int bucket = HighestBitSet(current_size + kFirstBucketCapacity) - HighestBitSet(kFirstBucketCapacity);
    if (data_[bucket].load(std::memory_order_acquire) == nullptr) {
        AllocBucket(bucket);
    }
    
    auto oldValue = At(current_size)->load(std::memory_order_acquire);
    auto write_op = internal::write_descriptor<T>(oldValue, value, current_size, false);
    new_descriptor = new internal::v_descriptor<T>(current_size + 1, write_op);
} while (!descriptor_.compare_exchange_strong(current_descriptor, new_descriptor, std::memory_order_acq_rel));

// 完成写操作并释放旧descriptor
CompleteWrite(new_descriptor->writeDescriptor_);
delete current_descriptor;

同时添加析构函数清理内存：

~LockFreeArray() {
    delete descriptor_.load();
    for (size_t i = 0; i < kNumberOfBuckets; ++i) {
        delete[] data_[i].load();
    }
    delete[] data_;
}

2. 修复CompleteWrite的参数传递问题 #

当前CompleteWrite按值传递write_descriptor，修改completed_仅作用于副本，无法同步到原对象。改为引用传递：

void CompleteWrite(internal::write_descriptor<T>& write_descriptor) {
    if (!write_descriptor.completed_) {
        std::atomic<T>* memoryLocation = At(write_descriptor.location_);
        // 指定内存序保证可见性
        if (memoryLocation->compare_exchange_weak(write_descriptor.old_value_, write_descriptor.new_value_,
                                                  std::memory_order_acq_rel, std::memory_order_acquire)) {
            write_descriptor.completed_ = true;
        }
    }
}

3. 修复Size()函数的线程安全性 #

当前Size()直接访问descriptor_->writeDescriptor_.completed_，未保证原子性。改为原子加载当前descriptor后再检查状态：

size_t Size() noexcept {
    auto current_descriptor = descriptor_.load(std::memory_order_acquire);
    if (!current_descriptor) return 0;
    
    size_t current_size = current_descriptor->size_;
    if (!current_descriptor->writeDescriptor_.completed_) {
        current_size--;
    }
    return current_size;
}

4. 修复AllocBucket的线程安全与内存序问题 #

AllocBucket中需指定内存序避免指令重排，且提前判断桶是否已分配，减少不必要的内存分配：

void AllocBucket(int bucket) {
    int bucketSize = 1 << (bucket + HighestBitSet(kFirstBucketCapacity));
    std::atomic<T>* currentBucket = data_[bucket].load(std::memory_order_acquire);
    if (currentBucket) return; // 已被其他线程分配，直接返回
    
    auto newBucket = new std::atomic<T>[bucketSize];
    if (!data_[bucket].compare_exchange_weak(currentBucket, newBucket,
                                             std::memory_order_acq_rel, std::memory_order_acquire)) {
        delete[] newBucket;
    }
}

5. 修复At()函数的边界检查 #

添加断言避免访问未分配的内存，防止堆损坏：

#include <cassert>

std::atomic<T>* At(size_t index) const {
    assert(index < Size()); // 检查索引是否超出有效范围
    auto pos = index + kFirstBucketCapacity;
    auto hibit = HighestBitSet(pos);
    int bucket_idx = hibit - HighestBitSet(kFirstBucketCapacity);
    assert(bucket_idx < kNumberOfBuckets);
    
    auto bucket_ptr = data_[bucket_idx].load(std::memory_order_acquire);
    assert(bucket_ptr != nullptr); // 确保桶已分配
    
    auto index_in_bucket = (pos ^ (1 << hibit));
    assert(index_in_bucket < (1 << hibit)); // 检查桶内索引边界
    return &bucket_ptr[index_in_bucket];
}

6. 内存管理优化 #

使用裸指针管理v_descriptor时，需避免内存泄漏。可引入epoch-based内存回收机制（无锁数据结构通用方案），或在确认所有线程不再访问旧descriptor后再释放内存。

编译选项优化 #

添加编译选项确保原子操作和内存序的正确性：

clang++ -std=c++20 -stdlib=libc++ -O2 -pthread your_code.cpp

内容的提问来源于stack exchange，提问作者Mihai Popescu