尝试使用“Readers Writer Lock”时的竞争条件

Question

我正在使用pthreads 进行一个项目，我自己实现了 Readers Writer Lock，它具有以下方法：

• 阅读器锁定（可同时阅读多个）
• 写者锁（只有一个人可以写）
• 解锁（读写器）

我已经测试过了，效果很好，我的问题更符合逻辑， 在我的程序中，我希望几个线程对特定范围内的数字进行一些测试，如果找到一个符合我的标准的数字，我希望他们将它们添加到一个共享资源中，该资源是一个数组。

如果该数字已被另一个线程找到并且存在于数组中，则继续搜索。

这是我的算法的伪代码：

X = lowest number to search, X' = highest number to search,
func = the test for the number, ARR = the array shared between the threads, 
RWL_R = Lock for reader, RWL_W Lock for writer, RWL_U = Unlock.


FOR each NUM from X to X' DO:
    IF func(NUM) = true DO:
        FOR each cell of ARR DO:
            RWL_R // lock for reader since i'm reading from ARR
            IF ARR[cell] contains NUM DO:
                RWL_U // unlock since no longer reading from ARR
                skip to the next iteration of the first FOR loop
            ENDIF
        END FOR

        RWL_U // unlock since no longer reading from ARR
        ////// PROBLEM HERE ////////
        RWL_W // lock to write to array since for loop ended without finding the same NUM

        ARR[cell] <- NUM
        RWL_U // unlock since finished writing into array
    END IF
END FOR

正如您所见，逻辑很好，因为我用丑陋的大写字母“PROBLEM HERE”标记了那条小线。在读取器解锁和写入器锁定之间的小间隙内，可能（并且确实）会发生竞争条件。

所以我有两个可能的结果：

1. （善）

   • Thread_A 找到数字 N，锁定数组以供读取。
   • Thread_B 找到相同的数字 N，等待检查数组，但它当前被 Thread_A 锁定。
   • Thread_A 完成遍历数组并且数字 N 不存在，解锁锁并作为写入器再次锁定它，将 N 添加到数组中，解锁锁并完成他的工作。
   • Thread_B 现在可以检查数组，编号 N 在那里，所以它跳到编号 N2，其余的工作正常。

2. （坏的）

   • Thread_A 找到数字 N，锁定数组以供读取。
   • Thread_B 找到相同的数字 N，等待检查数组，但它当前被 Thread_A 锁定。
   • Thread_A 完成遍历数组并且数字 N 不存在，解锁锁。
   • Thread_B 接管锁并将其锁定为读取器，检查数组和数字 N 仍然不存在（Thread_A 尚未添加它）。

   • Thread_B 解锁。

   • 现在 Thread_A 或 Thread_B 锁定写操作的锁，添加数字 N，解锁锁并完成。

   • 等待的线程现在锁定锁，添加相同的数字 N，解锁并完成。

所以我现在正试图找到解决问题的最佳逻辑方法，我只能在检查数组时考虑锁定作为写入器，直到完成写入才解锁它，或者创建一个“原子地切换”的方法” 从读取器锁到写入器锁，但这是一种“作弊”，而不是使用应该使用的“Readers Writer Lock”机制。

在这里使用它有什么更好的逻辑方式？

Answer 1

在大多数情况下，您给出的两个选项都不是最佳选择，因为一个选项可以防止多个阅读器同时检查（并且大概过了一段时间，阅读器比编写器更常见），而另一个是不可能的；即使您“原子地”从读取器锁切换到写入器锁，两个不同的读取器都可以确定值 X 不存在，都请求升级写入锁，第一个写入 X 然后解锁，而第二个等待轮到它再次写 X，你最终违反了唯一性不变量。

真正的答案取决于常见的使用模式：

1. 如果通常写入比读取更可能发生（并发读取并不常见），那么您只需删除读取器/写入器锁并使用简单的互斥锁即可。如果您不经常使用并发读取器，即使开销很小（有时是；Windows 的 SRW Locks 仅在“独占”（写入器）模式下使用），读取器/写入器锁通常会产生不值得的开销不需要递归获取的地方与临界区一样快或更快），如果程序逻辑变得更复杂，或者您需要不断地从读锁切换到写锁并返回，那么额外的逻辑和获取/释放开销可能会花费更多而不仅仅是一个锁定和独占访问解锁。

2. 如果您有其他经常使用的代码读取但从不写入，并且可能需要编写的阅读器不太常见并且经常需要编写，那么使用您建议的解决方案；保持读/写锁，但从一开始就有写“可能写”的代码锁；它消除了“可能写”代码路径中的并发，但“只读”用户可以并发操作。

3. 如果读取比写入更常见（如果提供的代码是访问数组的唯一代码，这是此类事情的常见用例；否则您的数组将失控），然后执行“升级”写锁后仔细检查；在升级到写锁后再次执行相同的扫描，以确保没有人抓住写锁并添加缺失值，然后如果该值仍然缺失则写入。如果数组仅添加了新值，并且现有值永远不会更改或被删除，则可以优化这一点以避免重新扫描。您只需要记住您停止检查的位置，并扫描在原始扫描和获得写锁之间添加的任何新条目。

#3 的伪代码如下所示：

FOR each NUM from X to X' DO:
    IF func(NUM) = true DO:
        RWL_R // lock for reader outside inner loop since repeatedly reading from ARR
        cell = 0
        WHILE cell < ARR.length DO:
            IF ARR[cell] contains NUM DO:
                RWL_U // unlock since no longer reading from ARR
                skip to the next iteration of the first FOR loop
            ENDIF
            cell += 1
        END WHILE

        RWL_U // unlock since no longer reading from ARR
        RWL_W // lock to write to array since for loop ended without finding the same NUM
        // NEW!!! Check newly added cells
        WHILE cell < ARR.length DO:
            IF ARR[cell] contains NUM DO:
                RWL_U // unlock since no longer reading from ARR
                skip to the next iteration of the first FOR loop
            ENDIF
            cell += 1
        END WHILE
        // If we got here, not in newly added cells either, so add to end
        ARR[cell] <- NUM
        RWL_U // unlock since finished writing into array
    END IF
END FOR

Answer 2

我只能在检查数组时考虑锁定为写入器，直到完成写入才解锁

这当然是可行的，但它会阻止阵列扫描的任何并发性。如果这些消耗了程序运行时间的很大一部分，那么使扫描并发是非常可取的。

或者创建一个从读取器锁“原子”切换到写入器锁的方法，但这是一种“作弊”，而不是使用应该使用的“读取器写入器锁”机制。

这是不可行的，因为当多个线程持有读锁时，您不能将读锁提升为写锁。写锁不仅必须排除其他写入者，还必须排除所有读取者。在您描述的情况下，您最终会遇到死锁，因为两个或多个持有读锁的线程需要将其提升为写锁，但在所有其他线程释放锁之前，它们中的任何一个都不会发生这种情况。

在任何情况下，即使您允许写锁与读锁共存，这也不能可靠地阻止考虑相同编号的两个线程都扫描文件到其当前结尾，而不是看到编号，反过来，将其附加到数组中。

---

如果您想提供并发数组扫描但防止将重复项添加到您的数组中，那么您至少需要多线程之间的通信。

一种相对简单的方法是实现一个简单的事务系统。您的读取器/写入器锁将支持在有多个读取器时将读取锁提升为写入锁，但仅适用于自上次释放写入器锁以来获得的读取器锁。成功执行这种提升的线程可以确信在读取数据时数据没有被修改，因此它可以安全地更新数组。提升失败类似于提交事务失败——遇到此类失败的线程需要从头开始重新扫描。

如果每个数字出现的次数相对较多，这将相当有效，因为重新扫描的可能性会随着重新扫描成本的增加而降低。

您还可以考虑更复杂的机制，其中获取写锁的线程以某种方式通知读者他们正在写入什么值，以便任何扫描该值的读者可以提前中止。然而，这样写起来会比较棘手，虽然听起来很高效，但在某些情况下，由于需要更多线程之间的同步，它实际上可能会更慢。

Answer 3

不要使用单个读取器/写入器锁，而是使用两个锁：一个用于读取，一个用于写入。每个线程将获取读锁并扫描数组。如果要修改数组，则线程将获取写锁并重新扫描数组以确保尚未添加数字。

这是伪代码：

acquire_read_lock();
if (!is_value_in_array(value)) {
    acquire_write_lock();
    if (!is_value_in_array(value)) {
        add_value_to_array(value);
    }
    release_write_lock();
}
release_read_lock();