使用Rust进行手动锁定

我正在尝试在Rust中编写union-find的实现.这在C语言中实现非常简单,同时仍然具有复杂的运行时分析.

我无法获得Rust的互斥锁语义以允许迭代的手动锁定.

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

首先,这是我在C中想要的部分结构的一个非常简单的实现:

#include 

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>
}

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

fn find_root(x: Rc) -> Rc {
    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当多个指向项目的指针处于活动状态时,接口只允许通过copy-on-write进行可变访问.因此,所有代码都必须改变.在进入路径压缩部分之前,我被挂断了find_root:

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

struct Node {
    parent: Option>>
}

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

fn find_root(x: Arc>) -> Arc> {
    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>而不是Arc>.这样做会导致相同的编译器错误.

为了清楚我要通过使用手动锁定来避免什么,这里是一个RefCell版本,编译但在路径的长度使用线性空间.

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

reem.. 8

我们可以很容易地完成手动锁定操作,因为我们只使用了一点点遍历这个列表unsafe,这对于告诉借用检查器我们知道的一小部分洞察力是必要的,但它无法知道.

但首先,让我们清楚地阐述问题:

我们想要遍历存储Arc>节点的链表,以获取列表中的最后一个节点

我们需要锁定列表中的每个节点,以便另一个并发遍历必须紧跟在我们后面,并且不能破坏我们的进度.

在我们深入了解细节之前,让我们尝试为此函数编写签名:

fn find_root(node: Arc>) -> Arc>;

现在我们知道了我们的目标,我们可以开始实施 - 这是第一次尝试:

fn find_root(incoming: Arc>) -> Arc> {
    // 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我们可以解决这个问题.

这里的关键见解是,a中包含的数据的生命周期Arc是动态的,并且借用检查器很难做出关于an内部数据的确切Arc有效时间的推断.

写锈的时候偶尔会发生这种情况; 你比生锈更了解数据的生命周期和组织,你希望能够向编译器表达这些知识,有效地说"信任我".输入:unsafe- 我们告诉编译器我们知道的比它更多的方式,它应该允许我们告知它我们知道的保证,但事实并非如此.

在这种情况下,保证非常简单 - 我们将在锁定仍然存在的情况下替换节点,但我们不会确保锁定内的数据继续有效,即使节点消失.为了表达这种保证,我们可以使用mem::transmute一个允许我们重新解释任何变量类型的函数,只需使用它来改变节点返回的锁的生命周期,使其略长于实际值.

为了确保我们保持我们的承诺,我们将使用另一种切换变量保存节点,而我们重新分配锁-即使该移动节点(改变其地址),并借检查将在我们愤怒,我们知道它的确定,因为lock没有按指向节点,它指向其中的数据node,其地址(在这种情况下,因为它在后面Arc)不会改变.

在我们开始解决之前,重要的是要注意我们在这里使用的技巧只有在我们使用的时候才有效Arc.借用检查员警告我们可能存在严重错误 - 如果Mutex是内联而不是在内Arc,则此错误将是正确防止免费使用后的情况,其中MutexGuard所持有的lock将尝试解锁Mutex已经存在的错误丢弃,或至少移动到另一个内存位置.

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

fn find_root(incoming: Arc>) -> Arc> {
    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很高兴,而且我们有手动锁定,因为最后一次锁定只有在我们获得新锁之后才会释放!

在这个实现中有一个未解决的问题,我还没有得到答案,即旧值的丢弃和新值的赋值是否保证是原子的 - 如果不是,那么就有一个种族在分配中获取新锁之前释放旧锁的条件lock.只需要另外一个holdover_lock变量并在重新分配之前将旧锁移动到其中,然后在重新分配后将其丢弃,这是非常简单的lock.

希望这完全解决了您的问题,并展示了unsafe当您真正了解更多信息时,如何利用借用检查器中的"缺陷"来解决问题.我仍然希望你知道的情况比借用检查器更为罕见,并且转换生命周期不是"通常"的行为.

使用Mutex这种方式,你可以看到,是很复杂的,你必须处理许多,许多的比赛条件,可能的来源,我可能甚至没有抓住所有的人!除非你真的需要这种结构是由许多线程访问,这很可能是最好只使用Rc和RefCell,如果你需要它,因为这使事情变得多容易.