如何定义线程安全数组？

Question

如何以最少的修改定义一个线程安全的全局数组？

我希望通过使用互斥锁和同步块来完成对它的每次访问。

像“T”这样的东西将是某种类型（请注意，AFAIK 当前未定义“同步”关键字）：

sync Array!(T) syncvar;

并且每次访问它都会与此类似：

Mutex __syncvar_mutex;

    //some func scope....
    synchronized(__syncvar_mutex) { /* edits 'syncvar' safely */ }

Answer 1

我天真的尝试是做这样的事情：

import std.typecons : Proxy:

synchronized class Array(T)
{
    static import std.array;
    private std.array.Array!T data;
    mixin Proxy!data;
}

很遗憾，由于https://issues.dlang.org/show_bug.cgi?id=14509，它不起作用

不能说我很惊讶，因为在现代 D 中通过隐藏的互斥锁自动处理多线程是非常单一的，并且同步类的概念主要是 D1 时代的遗留物。

当然，您可以手动实现相同的解决方案，方法是定义自己的 SharedArray 类和所有必要的方法，并在调用内部私有纯 Array 方法之前在方法内添加锁。但我想你想要一些开箱即用的东西。

现在不能发明任何更好的东西（会考虑更多）但值得注意的是，通常在 D 中鼓励创建旨在明确处理共享访问的数据结构，而不是仅仅保护正常的数据结构与互斥锁。当然，最受鼓励的方法是完全不使用消息传递来共享数据。

如果我有更好的想法，我会更新答案。

Answer 2

在数组周围创建一个包装器是相当容易的，这将使它成为线程安全的。但是，要制作一个没有并发瓶颈的线程安全数组是极其困难的。

想到的最接近的是 Java 的 CopyOnWriteArrayList 类，但即使这样也不理想......

Answer 3

您可以将数组包装在一个 struct 中，当线程获取令牌时锁定对数组的访问，直到它释放它为止。

包装器/储物柜：

• acquire(): 被线程循环调用。因为它返回一个指针，当方法返回一个非空值时，线程就知道它拥有令牌。
• release()：在处理了之前已经获得访问权的数据后被线程调用。

.

shared struct Locker(T)
{
    private:
        T t;
        size_t token;   
    public:  
        shared(T) * acquire() 
        {
            if (token) return null;
            else
            {
                import core.atomic;
                atomicOp!"+="(token, 1);
                return &t;
            }
        }
        void release()
        {
            import core.atomic;
            atomicOp!"-="(token, 1);
        }
}

快速测试：

alias LockedIntArray = Locker!(size_t[]);
shared LockedIntArray intArr;

void arrayTask(size_t cnt)
{
    import core.thread, std.random;

    // ensure the desynchronization of this job.
    Thread.sleep( dur!"msecs"(uniform(4, 20)));

    shared(size_t[])* arr = null;
    // wait for the token
    while(arr == null) {arr = intArr.acquire;}

    *arr ~= cnt;    
    import std.stdio;
    writeln(*arr);

    // release the token for the waiting threads
    intArr.release;
}

void main(string[] args)
{
    import std.parallelism;
    foreach(immutable i; 0..16)
    {
       auto job = task(&arrayTask, i);
       job.executeInNewThread(); 
    } 
}

缺点是对数组的每个操作块都必须用 acquire/release 对包围。

Answer 4

你的想法是对的。作为一个数组，您需要能够编辑和检索信息。我建议你看看 Phobos 提供的 read-write mutex 和 atomic 实用程序。读取操作相当简单：

1. synchronize mutex.readLock
2. 加载（使用atomicLoad）
3. 将项目从synchronize 块中复制出来
4. 返回复制的项目

写作应该几乎完全相同。只需在mutex.writeLock 上进行syncronize 并执行cas 或atomicOp 操作。

请注意，这仅在您在读取期间复制数组中的元素时才有效。如果要获取引用，每次访问或修改元素时都需要对元素做额外的同步。