【问题标题】:Hand-over-hand locking with Rust使用 Rust 进行手动锁定
【发布时间】:2014-12-06 23:53:35
【问题描述】:

我正在尝试用 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 接口只允许在多个时通过写时复制进行可变访问指向该项目的指针是活动的。结果,所有代码都必须更改。在进入路径压缩部分之前,我就挂断了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,...当然,我可以保持所有锁打开:lock A,锁 B,锁 C,...解锁 C,解锁 B,解锁 A,但这似乎效率低下。

但是,Mutex 不提供解锁功能,而是使用 RAII。 如果不能直接调用 unlock,如何在 Rust 中实现手动锁定?


编辑:正如 cmets 所说,我可以使用 Rc&lt;RefCell&lt;Node&gt;&gt; 而不是 Arc&lt;Mutex&lt;Node&gt;&gt;。这样做会导致相同的编译器错误。

为了清楚说明我试图通过使用手动锁定来避免什么,这里有一个 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()
    }
}

【问题讨论】:

  • 不,Mutex 在这里几乎肯定是错误的选择。如果您停留在单个任务中,您可能应该(默认情况下),最好使用Rc&lt;Cell&lt;T&gt;&gt;(如果TCopy)或Rc&lt;RefCell&lt;T&gt;&gt;(如果不是)来改变Rc 后面的内容.
  • 我同意CellRefCell 是正确的途径。您有一个看起来不可变的方法 (find),但想要在幕后改变数据。这是 Cell 家族的主要案例。
  • 当使用Rc&lt;RefCell&lt;Node&gt;&gt; 代替Arc&lt;Mutex&lt;Node&gt;&gt;(和borrow_mut 代替lock)时,我得到完全相同的错误:“错误:无法分配给ans,因为它借来的……”

标签: data-structures locking rust


【解决方案1】:

当我们使用unsafe 遍历这个列表时,我们可以很容易地完成完全的手动锁定,这对于告诉借用检查器一些我们知道的洞察力是必要的,但是它不知道。

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

  • 我们要遍历一个链表,其节点存储为Arc&lt;Mutex&lt;Node&gt;&gt;,得到链表中的最后一个节点
  • 我们需要在前进的过程中锁定列表中的每个节点,以便另一个并发遍历必须严格跟在我们后面,并且不能影响我们的进度。

在我们进入细节之前,让我们尝试为这个函数编写签名:

fn find_root(node: Arc&lt;Mutex&lt;Node&gt;&gt;) -&gt; Arc&lt;Mutex&lt;Node&gt;&gt;;

既然我们知道了我们的目标,我们就可以开始着手实施了——这是第一次尝试:

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,这个函数允许我们重新解释任何变量的类型,只需使用它来将节点返回的锁的生命周期更改为比实际更长一点。

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

在我们找到解决方案之前,请务必注意,我们在此处使用的技巧有效,因为我们使用的是Arc。借用检查器警告我们可能存在严重错误 - 如果 Mutex 被内联而不是在 Arc 中,则此错误将正确防止 use-after-free,其中 MutexGuardlock 将尝试解锁已被删除或至少移动到另一个内存位置的 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().unwrap();
    
    // 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().unwrap()) };
    }

    node
} 

而且,就像这样,rustc 很高兴,我们有手交锁,因为最后一个锁只有在我们获得新锁之后才会释放!

在这个实现中有一个未解决的问题,我也没有收到答案,那就是删除旧值和为变量分配新值是否保证是原子的 - 如果不是,那么是在分配lock 中获取新锁之前释放旧锁的竞争条件。只需使用另一个 holdover_lock 变量并在重新分配之前将旧锁移入其中,然后在重新分配 lock 后将其删除,就可以轻松解决这个问题。

希望这完全解决了您的问题,并展示了当您确实了解更多信息时,如何使用unsafe 解决借用检查器中的“缺陷”。我仍然希望您知道的比借用检查器更多的情况很少见,并且转换生命周期不是“通常”的行为。

如您所见,以这种方式使用 Mutex 非常复杂,您必须处理 manymany、可能的竞争条件来源和我什至可能都没有抓住他们!除非您确实需要从多个线程访问此结构,否则最好只使用RcRefCell,如果您需要的话,因为这会使事情更加更容易。 p>

【讨论】:

  • 很好的答案!如果没有 unsafe 块,Mutex/MutexGuard 必须使用什么接口才能实现这一点?
  • Mutex 确实不可能公开一个允许这样做的接口,而无需以某种方式直接与Arc 集成。这里的问题是我们依赖于 Arc 的不变量,它在借用系统中无法表达,因为无法静态判断两个 Arcs 是否实际相关,并且只能安全地扩展如果它们指向相同的数据,则借用一个到另一个的生命周期。
【解决方案2】:

我认为这符合hand-over-hand locking 的标准。

use std::sync::Mutex;

fn main() {
    // Create a set of mutexes to lock hand-over-hand
    let mutexes = Vec::from_fn(4, |_| Mutex::new(false));

    // Lock the first one
    let val_0 = mutexes[0].lock();
    if !*val_0 {
        // Lock the second one
        let mut val_1 = mutexes[1].lock();
        // Unlock the first one
        drop(val_0);
        // Do logic
        *val_1 = true;
    }

    for mutex in mutexes.iter() {
        println!("{}" , *mutex.lock());
    }
}

编辑#1

当对锁n+1的访问被锁n保护时是否有效?

如果您的意思是可以形成如下形状,那么我认为答案是否定的。

struct Level {
  data: bool,
  child: Option<Mutex<Box<Level>>>,
}

但是,这不应该工作是明智的。当您将对象包装在互斥体中时,您就是在说“整个对象是安全的”。你不能同时说“整个派是安全的”和“我正在吃地壳下面的东西”。也许您通过创建 Mutex&lt;()&gt; 并锁定它来抛弃安全?

【讨论】:

  • 手动锁定是否循环工作?当对锁 n+1 的访问被锁 n 保护时是否有效?
  • 也许这“不应该工作”,但我相信这是在并发链表上完成更新的标准方式,尽管保护(锁定 a 保护节点 b 的底部)是隐式的。在这些情况下,我相信互斥锁不是表示“通过这个锁访问的所有东西都是通过锁定这个互斥锁来获得的”,而是“通过这个锁访问的所有东西都没有获得任何其他锁是通过锁定这个互斥锁来获得的”。像您建议的那样走 Mutex 路线会留下数据竞争的可能性。
  • “标准方式”不一定适用于具有不同范式的语言^_^。 Rust 也有可能无法在语言的安全部分表达您的特定概念。也许您正在寻求的解决方案必须用暴露安全(因为程序员而不是编译器已验证)API 的不安全代码编写。
【解决方案3】:

这仍然不是您关于如何进行手动锁定的字面问题的答案,这应该只在并发设置中很重要(或者如果其他人强迫您使用 Mutex 对节点的引用)。而是如何使用您似乎感兴趣的 RcRefCell 来执行此操作。

RefCell 仅在持有一个可变引用时才允许可变写入。重要的是,Rc&lt;RefCell&lt;Node&gt;&gt; 对象是可变引用。它所说的可变引用是在Rc&lt;RefCell&lt;Node&gt;&gt;object 上调用borrow_mut() 的结果,只要您在有限的范围内(例如while 循环的主体)执行此操作,就可以了。

路径压缩中发生的重要事情是next Rc 对象将在您将node 的父指针指向root 时使链的其余部分保持活动状态。但是,它不是 Rust 意义上的 reference

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

fn find_root(mut node: Rc<RefCell<Node>>) -> Rc<RefCell<Node>>
{
    while let Some(parent) = node.borrow().parent.clone()
    {
        node = parent;
    }

    return node;
}

fn path_compress(mut node: Rc<RefCell<Node>>, root: Rc<RefCell<Node>>)
{
    while node.borrow().parent.is_some()
    {
        let next = node.borrow().parent.clone().unwrap();
        node.borrow_mut().parent = Some(root.clone());
        node = next;
    }
}

使用我使用的测试工具,这对我来说运行良好,尽管可能仍然存在错误。由于尝试borrow_mut() 已经借用的东西,它当然可以在没有panic! 的情况下编译和运行。它实际上可能会产生正确的答案,这取决于您。

【讨论】:

    【解决方案4】:

    在 IRC 上,Jonathan Reem 指出 inner 一直在借用到其词法范围的末尾,这对于我的要求来说太远了。内联它会产生以下内容,编译时不会出错:

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

    编辑: 正如 Francis Gagné 指出的那样,这有一个竞争条件,因为锁的延伸时间不够长。这是一个修改后的版本,只有一个 lock() 调用;也许它不容易受到同样问题的影响。

    fn find_root(x: Arc<Mutex<Node>>) -> Arc<Mutex<Node>> {
        let mut ans = x.clone();
        loop {
            ans = {
                let tmp = ans.lock();
                match tmp.parent.clone() {
                   None => break,
                   Some(z) => z
                }
            }
        }
        ans
    }
    

    编辑 2:这一次只持有一个锁,因此也是如此。我仍然不知道如何进行手动锁定。

    【讨论】:

    • 对于Mutex,这是一种危险的使用模式:该值可能会在第一次解锁和第二次锁定之间从另一个任务中改变。不过对于RefCell,这种模式很好,因为不存在数据竞争的风险。
    • @FrancisGagné,说得好。我将对此答案进行编辑,指出这一点;希望您能告诉我您对我将在编辑中添加的其他解决方案的看法。
    • 我将删除match 上的.clone() 调用并将Some(z) =&gt; z 更改为Some(ref z) =&gt; z.clone();它避免在None 时克隆Option
    【解决方案5】:

    正如 Frank Sherry 和其他人所指出的,单线程时不应使用 Arc/Mutex。但是他的代码已经过时了,所以这里是新的(对于版本 1.0.0alpha2)。 这也不占用线性空间(如问题中给出的递归代码)。

    struct Node {
        parent: Option<Rc<RefCell<Node>>>
    }
    
    fn find_root(node: Rc<RefCell<Node>>) -> Rc<RefCell<Node>> {
        let mut ans = node.clone(); // Rc<RefCell<Node>>
        loop {
            ans = {
                let ans_ref = ans.borrow(); // std::cell::Ref<Node>
                match ans_ref.parent.clone() {
                    None => break,
                    Some(z) => z
                }
            } // ans_ref goes out of scope, and ans becomes mutable
        }
        ans
    }
    
    fn path_compress(mut node: Rc<RefCell<Node>>, root: Rc<RefCell<Node>>) {
        while node.borrow().parent.is_some() {
            let next = {
                let node_ref = node.borrow();
                node_ref.parent.clone().unwrap()
            };
            node.borrow_mut().parent = Some(root.clone());
            // RefMut<Node> from borrow_mut() is out of scope here...
            node = next; // therefore we can mutate node
        }
    }
    

    初学者注意:指针由点运算符自动取消引用。 ans.borrow() 实际上是指(*ans).borrow()。我特意为这两个函数使用了不同的样式。

    【讨论】:

      【解决方案6】:

      虽然不是您的字面问题(移交锁定)的答案,但在 Rust 中使用加权联合和路径压缩的联合查找可以非常简单:

      fn unionfind<I: Iterator<(uint, uint)>>(mut iterator: I, nodes: uint) -> Vec<uint>
      {
          let mut root = Vec::from_fn(nodes, |x| x);
          let mut rank = Vec::from_elem(nodes, 0u8);
      
          for (mut x, mut y) in iterator
          {
              // find roots for x and y; do path compression on look-ups
              while (x != root[x]) { root[x] = root[root[x]]; x = root[x]; }
              while (y != root[y]) { root[y] = root[root[y]]; y = root[y]; }
      
              if x != y
              {
                  // weighted union swings roots
                  match rank[x].cmp(&rank[y])
                  {
                      Less    => root[x] = y,
                      Greater => root[y] = x,
                      Equal   =>
                      {
                          root[y] = x; 
                          rank[x] += 1 
                      },
                  }
              }
          }
      }
      

      也许元点是联合查找算法可能不是处理节点所有权的最佳位置,并且通过使用对现有内存的引用(在这种情况下,仅使用节点的 uint 标识符)而不影响节点的生命周期使得实现更加简单,当然如果你能摆脱它的话。

      【讨论】:

      • 这不允许被遗忘的节点被释放,直到整个 arena 被释放。它也没有回答在其他情况下(例如并发链表)中的手动锁定问题。
      • 好吧,您提供的 C 代码也没有。很难判断您的问题是关于解决问题(联合查找)还是探索编程模式。你现在说得更清楚了。
      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2014-05-01
      • 1970-01-01
      • 1970-01-01
      • 2016-09-23
      相关资源
      最近更新 更多