固定大小的并发列表

Question

我已经实现了一个简单的任务来创建一个固定大小的列表，该列表允许并发写入，并且可以随时转储列表中项目的最新快照。

这是我的实现。每个线程的偏移量将自动增加，如果达到列表的大小，则重置。不同的线程应该对数组的每个部分具有独立的访问权限。

我的问题是当我调用 Dump() 时，前几个项目没有存储在列表中。另外，是否有一个可以同时进行原子增加和重置的互锁函数，所以我不必创建一个更衣室对象和一个锁块？谢谢。

public static void Main(string[] args)
{            
    ConcurrentCircularFixedList<int> list = new ConcurrentCircularFixedList<int>(20);
    Enumerable.Range(1, 30).AsParallel().Select(nu => list.Enqueu(nu)).ToList();    
}

public class ConcurrentCircularFixedList<T>
{
    private int _size;
    private int _offset;
    private sealed object _locker = new Object();
    privateT[] _list;

    public ConcurrentCircularFixedList(int size)
    {
        _size = size;
        _offset = 0;
        _list = new T[_size];
    }

    public int Enqueu(T item)
    {
        _list[_offset] = item;
        lock(_locker)
        {
            Debug.Write("B " + _offset);
            _offset += 1;
            if(_offset == _size)
                _offset = 0;
            Debug.Write("A " + _offset + "\n");
         }  
         return _offset;
    }

    public T[] Dump()
    {
        return _list.ToArray();
    }
}

Answer 1

这是一个小版本的无锁列表，可在写入时复制。使用前应清楚了解其性能特点。当您有很多作家或列表很大时，它会很昂贵。由于列表实际上是不可变的，因此读取无需同步。当然，这可以通过各种方式改进，但你明白了。实际上，它牺牲了一些内存压力和较慢的写入速度来实现零成本读取。

public class CopyWriteList<T>
{
    private volatile List<T> list;

    public CopyWriteList()
    {
        list = new List<T>();
    }

    public CopyWriteList(int capacity)
    {
        list = new List<T>(capacity);
    }

    public T this[int index]
    {
        get { return list[index]; }
        set { Replace(x => x[index] = value); }
    }

    public void Clear()
    {
        Replace(x => x.Clear());
    }

    public void Add(T item)
    {
        Replace(x => x.Add(item));
    }

    //Etc....

    private void Replace(Action<List<T>> action)
    {
        List<T> current;
        List<T> updated;
        do
        {
            current = list;
            updated = new List<T>(current);
            action(updated);
        } while (Interlocked.CompareExchange(ref list, updated, current) != current);
    }

    public List<T> GetSnapshot()
    {
        return list;
    }
}

或者，这是您的代码的固定版本。请注意，读者和作者之间存在额外的争用。性能可能会因此受到影响（例如昂贵的上下文切换）。

public class ConcurrentCircularFixedList<T>
{
    private readonly int _size;
    private int _offset;
    private readonly object _locker = new Object();
    private readonly T[] _list;

    public ConcurrentCircularFixedList(int size)
    {
        _size = size;
        _offset = 0;
        _list = new T[_size];
    }

    public int Enqueue(T item)
    {
        lock (_locker)
        {
            _list[_offset] = item;
            Debug.Write("B " + _offset);
            _offset += 1;
            if (_offset == _size)
                _offset = 0;
            Debug.Write("A " + _offset + "\n");
            return _offset;
        }
    }

    public T[] Dump()
    {
        lock (_locker)
            return _list.ToArray();
    }
}