如何防止用餐哲学家c ++中的死锁

Posted 2021-04-29

我正试图解决用餐哲学家问题的僵局。我已经有了一个由我的老师提供的代码框架。

我尝试使用try_lock（）解决问题

chopstick[(i+1)%5].try_lock(); 

但这并没有解决我的问题，当我运行它时，我确实得到以下错误消息：错误“解锁无主互斥锁”。

我还尝试通过在youtube视频中看到的以下更改来解决问题

chopstick[i].lock();

至

chopstick[min(i,(i+1)%5)].lock();

并且

chopstick[(i+1)%5].lock();

至

chopstick[max(i,(i+1)%5)].lock();

这是我提供的骨架。

#include <windows.h>
#include <stdio.h>
#include <iostream>
#include <vector>
#include <algorithm>
#include <thread>
#include <mutex>
#include <time.h>

using namespace std;

thread task[5];
mutex chopstick[5];
int stop = false;

void go(int i) {
    while (!stop) {

        chopstick[i].lock();

        cout << i << ": takes: " << i << endl;
        chrono::milliseconds dur(20);
        this_thread::sleep_for(dur); //Leads to deadlock immediately
        chopstick[(i + 1) % 5].lock();

        cout << i << ": eating" << endl;

        chrono::milliseconds dur2(rand() % 200 + 100);
        this_thread::sleep_for(dur2);

        chopstick[(i + 1) % 5].unlock();
        chopstick[i].unlock();
    }
}
int main() {
    srand(time(NULL));

    for (int i = 0; i < 5; ++i) {
        task[i] = (thread(go, i));
    }
    for (int i = 0; i < 5; i++) {
        task[i].join();
    }

}

我在理论上理解餐饮哲学家，但我无法解决这个问题。我真的不明白我做错了什么。有人可以解释一下我做错了什么并帮我解决了吗？

答案

解决死锁的最简单方法是使用专为此目的而发明的std::lock(l1, l2)。

更改：

    chopstick[i].lock();

    cout << i << ": takes: " << i << endl;
    chrono::milliseconds dur(20);
    this_thread::sleep_for(dur); //Leads to deadlock immediately
    chopstick[(i + 1) % 5].lock();

至：

    std::lock(chopstick[i], chopstick[(i + 1) % 5]);

    cout << i << ": takes: " << i << endl;

这是一个直接的解决方案，它忽略了异常安全性，这对于让你的第一次死锁避免课程失败来说是好的。

为了使其异常安全，需要将互斥锁包装在RAII设备中：std::unique_lock：

    unique_lock<mutex> left{chopstick[i], defer_lock};
    unique_lock<mutex> right{chopstick[(i + 1) % 5], defer_lock};
    lock(left, right);

    cout << i << ": takes: " << i << endl;

然后你也应该删除明确的unlock语句，因为left和right的析构函数将处理这个问题。

现在，如果有任何东西在go中引发异常，left和right的析构函数将在异常传播时解锁互斥锁。

要了解有关std::lock引擎盖下发生的事情以避免死锁的更多信息，请参阅：http://howardhinnant.github.io/dining_philosophers.html

性能测试

这是一个快速简便的测试，用于比较std::lock的使用与更多传统的“命令你的互斥体”的建议。

#ifndef USE_STD_LOCK
#   error #define USE_STD_LOCK as 1 to use std::lock and as 0 to use ordering
#endif

#include <atomic>
#include <chrono>
#include <exception>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <thread>

std::thread task[5];
constexpr auto N = sizeof(task)/sizeof(task[0]);
std::mutex chopstick[N];
std::atomic<bool> stop{false};
unsigned long long counts[N] = {};

using namespace std::chrono_literals;

void
go(decltype(N) i)
{
    auto const right = (i + 1) % N;
    decltype(right) const left = i;
    while (!stop)
    {
#if USE_STD_LOCK
        std::lock(chopstick[left], chopstick[right]);
#else
        if (left < right)
        {
            chopstick[left].lock();
            chopstick[right].lock();
        }
        else
        {
            chopstick[right].lock();
            chopstick[left].lock();
        }
#endif
        std::lock_guard<std::mutex> l1{chopstick[left],  std::adopt_lock};
        std::lock_guard<std::mutex> l2{chopstick[right], std::adopt_lock};
        ++counts[i];
        std::this_thread::sleep_for(1ms);
    }
}

void
deadlock_detector(std::chrono::seconds time_out)
{
    std::this_thread::sleep_for(time_out);
    std::cerr << "Deadlock!
";
    std::terminate();
}

int
main()
{
    for (auto i = 0u; i < N; ++i)
        task[i] = std::thread{go, i};
    std::thread{deadlock_detector, 15s}.detach();
    std::this_thread::sleep_for(10s);
    stop = true;
    for (auto& t : task)
        t.join();
    std::cout << std::right;
    for (auto c : counts)
        std::cout << std::setw(6) << c << '
';
    auto count = std::accumulate(std::begin(counts), std::end(counts), 0ULL);
    std::cout << "+ ----
";
    std::cout << std::setw(6) << count << '
';
}

必须使用USE_STD_LOCK定义编译：

1. #define USE_STD_LOCK 0订购你的互斥锁并按顺序锁定它们。
2. #define USE_STD_LOCK 1用std::lock锁定你的互斥体。

该程序运行10秒，每个线程尽可能频繁地增加一个不同的unsigned long long。但是为了让事情变得更加戏剧性，每个线程也会在持有锁的同时休眠1ms（如果你愿意，可以在没有睡眠的情况下运行）。

在10s之后，main告诉每个人轮班结束并计算每个线程的结果，以及所有线程的总增量。越高越好。

在启用优化的情况下运行，我得到如下数字：

USE_STD_LOCK = 1

  3318
  2644
  3254
  3004
  2876
+ ----
 15096

USE_STD_LOCK = 0

    19
    96
  1641
  5885
    50
+ ----
  7691

请注意，使用std::lock不仅会导致更高的累积结果，而且每个线程对总数的贡献大致相同。相反，“排序”倾向于选择单个线程，在某些情况下使其他线程挨饿。

这是一款4核的英特尔酷睿i5。我将差异归结为具有多个核心，以便至少两个线程可以并发运行。如果这是在单个核心设备上运行，我不会期望这种差异（我没有测试过该配置）。

我还使用死锁检测器对测试进行了检测。这不会影响我得到的结果。它旨在让人们尝试其他锁定算法，并更快地确定测试是否已锁定。如果这个死锁检测器以任何方式困扰您，只需将其从测试运行中删除即可。我不想辩论它的优点。

如果你得到类似的结果或不同，我欢迎建设性的反馈。或者，如果您认为此测试偏向于某种方式，以及如何使其更好。

另一答案

并行编程（带锁）的核心规则之一是，您应该始终以相同的顺序获取锁。

在您的代码中，每个任务首先获取其锁定，然后执行下一个锁定。一种解决方案是始终从偶数索引中获取锁定，然后才从奇数索引中获取锁定。这样，您获取锁定的顺序将保持一致。

另一个众所周知的策略是'退避'，你使用lock()获取第一个锁，然后使用try_lock()获取后续锁，如果无法获得，则释放所有获得的锁并重新启动序列。这种策略在性能方面并不好，但它保证最终会起作用。

另一答案

有四（4）个条件是必要的，足以产生死锁。

死锁条件

• 资源互斥（资源无法共享）

考虑（请求）的资源不得分享。当允许共享资源时，不会阻止（兄弟）进程在需要时获取资源。

• 资源保持和等待或持久性（部分分配）

进程必须保留已分配的资源，并等待（尝试占用）后续（请求的）资源。当进程必须在请求新资源时释放所持有的资源时，则不会发生死锁，因为进程不会阻止（兄弟）进程在需要时获取资源。

• 不允许资源抢占（流程平等或公平）

进程不能在持有时带走资源。否则，更高优先级（排名）过程将简单地占用（抓住）足够的资源以使过程完成。许多RTOS使用此方法来防止死锁。

• 资源循环订单或等待（资源图中存在循环）

资源中存在循环排序或循环（链），其中资源不能按部分顺序排列（编号为min .. max）。当可以对资源施加部分顺序时，可以按照严格的顺序来获取（锁定）资源，并且不会发生死锁（参见循环定理，其中指出“资源图中的循环是必要的，以便死锁可以发生“）。

Dining Philosophers问题（虽然是实验）被构建为呈现所有四个条件，并且挑战在于决定避免（中断）哪些条件。一个经典的答案是改变资源的顺序以打破循环等待条件。每个哲学家都独立决定如何解决僵局。

• 可分享 - 每个哲学家需要两个叉子，不能分享。
• 坚持不懈 - 每个哲学家必须保留一个叉子而另一个。
• 没有先发制人 - 没有哲学家会从另一个人那里获得一个分支。
• 循环秩序 - 有一个循环，所以两个哲学家碰撞和僵局。

有几个众所周知的解决方案：

• Dijkstra的解决方案 - 对叉子（1 .. N）进行编号，所有哲学家都根据规则采取分叉：采用较低编号的叉子，然后使用较高编号的叉子，并在碰撞时释放任何资源。
• 仲裁员（监督员） - 仲裁员分配分叉，当哲学家想吃东西时，他们会询问仲裁员，他们将他们的请求序列化，并在可用时分配（分发）资源（分叉）。

Dijkstra是规范的解决方案 - 为您的叉号编号。