data-structures - 用Rust进行交接锁定

Question

我正在尝试在Rust中编写union-find的实现。众所周知，这非常容易用C之类的语言实现，同时仍具有复杂的运行时分析。

我在获取Rust的互斥体语义以允许迭代的跨手锁定方面遇到麻烦。

这就是我现在所处的位置。

首先，这是我想要用C编写的结构的一部分的非常简单的实现:

#include <stdlib.h>

struct node {
  struct node * parent;
};

struct node * create(struct node * parent) {
  struct node * ans = malloc(sizeof(struct node));
  ans->parent = parent;
  return ans;
}

struct node * find_root(struct node * x) {
  while (x->parent) {
    x = x->parent;
  }
  return x;
}

int main() {
  struct node * foo = create(NULL);
  struct node * bar = create(foo);
  struct node * baz = create(bar);
  baz->parent = find_root(bar);
}

请注意，指针的结构是 倒置树的结构；多个指针可能指向单个位置，并且没有循环。

此时，没有路径压缩。

这是Rust的翻译。我选择使用Rust的引用计数指针类型来支持我上面引用的倒置树类型。

请注意，此实现更为冗长，这可能是由于Rust提供的安全性提高，但可能是由于我对Rust缺乏经验。

use std::rc::Rc;

struct Node {
    parent: Option<Rc<Node>>
}

fn create(parent: Option<Rc<Node>>) -> Node {
    Node {parent: parent.clone()}
}

fn find_root(x: Rc<Node>) -> Rc<Node> {
    let mut ans = x.clone();
    while ans.parent.is_some() {
        ans = ans.parent.clone().unwrap();
    }
    ans
}

fn main() {
    let foo = Rc::new(create(None));
    let bar = Rc::new(create(Some(foo.clone())));
    let mut prebaz = create(Some(bar.clone()));
    prebaz.parent = Some(find_root(bar.clone()));
}

每次调用 find_root时，路径压缩都会重新定向到根的路径上的每个节点。要将此功能添加到C代码中，仅需要两个新的小功能:

void change_root(struct node * x, struct node * root) {
  while (x) {
    struct node * tmp = x->parent;
    x->parent = root;
    x = tmp;
  }
}

struct node * root(struct node * x) {
  struct node * ans = find_root(x);
  change_root(x, ans);
  return ans;
}

函数 change_root进行所有重新父操作，而函数 root只是使用 find_root的结果来重新父级到根路径上的节点的包装器。

为了在Rust中执行此操作，我决定我必须使用 Mutex而不是引用计数的指针，因为 Rc接口(interface)仅在有多个指向该项目的指针存在时才允许通过写时复制进行可变访问。结果，所有代码都必须更改。甚至在进入路径压缩部分之前，我都卡在 find_root上:

use std::sync::{Mutex,Arc};

struct Node {
    parent: Option<Arc<Mutex<Node>>>
}

fn create(parent: Option<Arc<Mutex<Node>>>) -> Node {
    Node {parent: parent.clone()}
}

fn find_root(x: Arc<Mutex<Node>>) -> Arc<Mutex<Node>> {
    let mut ans = x.clone();
    let mut inner = ans.lock();
    while inner.parent.is_some() {
        ans = inner.parent.clone().unwrap();
        inner = ans.lock();
    }
    ans.clone()
}

这将产生错误(0.12.0)

error: cannot assign to `ans` because it is borrowed
ans = inner.parent.clone().unwrap();

note: borrow of `ans` occurs here
let mut inner = ans.lock();

我想我在这里需要的是交接锁定。对于路径A-> B-> C-> ...，我需要锁定A，锁定B，解锁A，锁定C，解锁B ... ...当然，我可以将所有锁定保持打开状态:锁定A，锁B，锁C，...解锁C，解锁B，解锁A，但这似乎效率很低。

但是， Mutex不提供解锁功能，而是使用RAII。 如何在Rust中实现交接锁定而又不能直接调用unlock？

编辑:如评论所述，我可以使用 Rc<RefCell<Node>>而不是 Arc<Mutex<Node>>。这样做会导致相同的编译器错误。

为了清楚起见，我尝试通过使用越区移交锁定来避免这种情况，这是一个 RefCell版本，可以编译但在路径长度上使用线性空间。

fn find_root(x: Rc<RefCell<Node>>) -> Rc<RefCell<Node>> {
    let mut inner : RefMut<Node> = x.borrow_mut();
    if inner.parent.is_some() {
        find_root(inner.parent.clone().unwrap())
    } else {
        x.clone()
    }
}

Answer 1

当我们仅使用unsafe遍历此列表时，我们可以很容易地进行完全交接锁定，这对于告诉借阅检查器我们知道的一小部分见解是必要的，但它不知道。

但是首先，让我们清楚地阐明问题:

我们要遍历一个链表，该链表的节点存储为Arc<Mutex<Node>>，以获取列表中的最后一个节点

我们在执行过程中需要锁定列表中的每个节点，以使另一个并发遍历必须严格遵循在我们身后，并且不能破坏我们的进度。

在深入了解细节之前，让我们尝试为该函数编写签名:
fn find_root(node: Arc<Mutex<Node>>) -> Arc<Mutex<Node>>;

既然我们知道了目标，就可以开始实现了-这是第一次尝试:

fn find_root(incoming: Arc<Mutex<Node>>) -> Arc<Mutex<Node>> {
    // We have to separate this from incoming since the lock must
    // be borrowed from incoming, not this local node.  
    let mut node = incoming.clone();
    let mut lock = incoming.lock();

    // Could use while let but that leads to borrowing issues.
    while lock.parent.is_some() {
       node = lock.parent.as_ref().unwrap().clone(); // !! uh-oh !!
       lock = node.lock();
    }

    node
} 

如果我们尝试对此进行编译，则rustc将在标记为 !! uh-oh !!的行上出错，告诉我们在 lock仍然存在的情况下我们无法移出节点，因为 lock借用了 node。这不是伪造的错误! lock中的数据可能会在 node中消失后立即消失-这仅仅是因为我们知道即使移动 lock可以解决此问题，我们也可以将 node指向的数据保持有效并位于同一内存位置。

此处的关键见解是 Arc中包含的数据的生命周期是动态的，并且借位检查器很难就 Arc中的数据有效时间进行推断。

在写 rust 时，这种情况偶尔发生。您比rustc更了解数据的生命周期和组织，并且希望能够向编译器表达这些知识，有效地说“信任我”。输入: unsafe-我们告诉编译器我们不仅仅了解的方法，它应该使我们可以将我们知道但不知道的保证告知它。

在这种情况下，保证非常简单-我们将在锁仍然存在的情况下替换节点，但是即使节点消失，我们也不会确保锁中的数据继续有效。为了表达这种保证，我们可以使用 mem::transmute，该函数允许我们重新解释任何变量的类型，只需使用它即可将节点返回的锁的生存期更改为比实际更长。

为了确保遵守 promise ，我们将在重新分配锁定时使用另一个切换变量来持有节点-即使这会移动节点(更改其地址)并且借用检查器会对我们生气，我们知道这是可以的，因为 lock不指向节点，而是指向 node内部的数据，该地址的地址(在这种情况下，因为它位于 Arc后面)不会改变。

在获得解决方案之前，必须注意，此处使用的技巧仅是有效的，因为我们使用的是 Arc。借用检查器会警告我们可能存在严重错误-如果 Mutex保持内联而不是 Arc中，则此错误将是对Free-After-Free的正确使用，因为 MutexGuard中包含的 lock会尝试解锁已被删除或至少已移动到另一个存储位置的 Mutex。

use std::mem;
use std::sync::{Arc, Mutex};

fn find_root(incoming: Arc<Mutex<Node>>) -> Arc<Mutex<Node>> {
    let mut node = incoming.clone();
    let mut handoff_node;
    let mut lock = incoming.lock();

    // Could use while let but that leads to borrowing issues.
    while lock.parent.is_some() {
       // Keep the data in node around by holding on to this `Arc`.
       handoff_node = node;

       node = lock.parent.as_ref().unwrap().clone();

       // We are going to move out of node while this lock is still around,
       // but since we kept the data around it's ok.
       lock = unsafe { mem::transmute(node.lock()) };
    }

    node
} 

而且，就像这样，rustc很高兴，而且我们拥有交接的锁定，因为只有在获得新的锁定之后才释放最后一个锁定!

此实现中有一个 Unresolved 问题，我也尚未得到答案，那就是是否一定要保证旧值的下降和将新值分配给变量是否可以保证是原子的-否则，就存在竞争在 lock的分配中，在获取新锁之前释放旧锁的情况。要解决此问题，只需再添加一个 holdover_lock变量，然后在重新分配之前将旧锁移入其中，然后在重新分配 lock之后将其删除，这是很简单的。

希望这可以完全解决您的问题，并说明当您确实了解更多信息时，如何使用 unsafe来解决借阅检查器中的“缺陷”。我仍然希望您很少了解借用检查器的情况，并且更改生命周期不是“通常”的行为。

如您所见，以这种方式使用 Mutex相当复杂，您必须处理许多很多可能的竞争条件源，而我什至没有捕获所有这些源!除非确实需要从多个线程访问此结构，否则最好使用 Rc和 RefCell，因为这样会使事情变得容易得多。