从 Rust 中的回调访问全局状态

Question

我正在学习 Rust，我一直在尝试用 Rust 重写我在 C# 中所做的一个项目，但我在尝试从回调访问全局状态时遇到了困难，

有没有一种简单的方法可以在 Rust 中做到这一点？请记住，我无法向回调添加新参数。

例如：

use std::collections::HashMap;
use std::time::instant;
use lib::bar;

struct Struct2{
    foo: bar,
    word: String,
}

struct GlobalState{
    running: bool,
    now: Instant,
    map: HashMap<String, Struct2>,
    hook_id: usize,
    current_id: String,
}

impl GlobalState{
    fn init() -> Self{
        let hook_id = unsafe {set_ext_hook(HOOK_ID, system_hook)};
        // Omitted initialization code.
    }
    // Omitted state mutation functions.
}

unsafe extern "system" fn system_hook(ext_param1:usize, ext_param2: usize) -> isize {
    // use global state here
}

我尝试使用 lazy_static 和 once_cell 等 crate，但它们不起作用，因为我使用的外部结构（本例中为 lib::bar） “无法在线程之间安全地发送”

到目前为止我的代码是单线程的（我计划在实现它时为程序的 gui 使用不同的线程）

感谢任何帮助，谢谢。

Answer 1

您似乎正在处理既不是 Send 也不是 Sync 的数据，因此 Rust 不允许您将它放在全局中，即使是在互斥体中。从问题中不清楚这是 lib::bar 真正线程不安全的结果，还是仅仅是其在幕后使用原始指针的意外结果。还不清楚您是否能够修改 lib::bar 使其类型为 Send 和 Sync.

最保守地假设 lib::bar 无法更改，并且考虑到您的程序是单线程的，您唯一安全的选择是创建线程本地状态：

use std::cell::RefCell;
use std::thread_local;

struct Foo(*const i32); // a non-Send/Sync type

struct GlobalState {
    foo: Foo,
    data: String,
    mutable_data: RefCell<String>,
}

thread_local! {
    static STATE: GlobalState = GlobalState {
        foo: Foo(std::ptr::null()),
        data: "bla".to_string(),
        mutable_data: RefCell::new("".to_string()),
    };
}

您可以从任何函数访问该状态（并修改其内部可变部分）：

fn main() {
    STATE.with(|state| {
        assert_eq!(state.foo.0, std::ptr::null());
        assert_eq!(state.data, "bla");
        assert_eq!(state.mutable_data.borrow().as_str(), "");
        state.mutable_data.borrow_mut().push_str("xyzzy");
    });
    STATE.with(|state| {
        assert_eq!(state.mutable_data.borrow().as_str(), "xyzzy");
    });
}

Playground

请注意，如果您尝试从不同的线程访问“全局”状态，每个线程都会获得自己的状态副本：

fn main() {
    STATE.with(|state| {
        state.mutable_data.borrow_mut().push_str("xyzzy");
    });
    std::thread::spawn(|| {
        STATE.with(|state| {
            // change to "xyzzy" happened on the other copy
            assert_eq!(state.mutable_data.borrow().as_str(), "");
        })
    })
    .join()
    .unwrap();
}

Playground

Answer 2

一个选项是变量的“线程限制”。这意味着对变量的所有访问都发生在一个线程上。通常，您为此创建一个专用线程，并为您的变量创建一个代理，该代理在其他线程之间共享，并负责从限制线程获取消息。

在 Rust 中，这是一种线程间通信，通常使用通道来完成。我将展示您的代码的简化版本——其中 lib::bar 只是包装了一个 i32 指针。指针不实现 Send+Sync，是 API.

的一个很好的替代品

代码相当冗长，我作弊并且没有对所有 send 和 recv 调用实施错误处理，您绝对应该这样做。尽管冗长，但添加新功能非常简单 - 它主要包括向 Message 和 Reply 枚举添加变体，并复制现有功能。

use lazy_static::lazy_static;
use std::sync::mpsc::sync_channel;

pub mod lib {
    pub struct Bar(*mut i32);
    impl Bar {
        pub fn new() -> Self {
            Bar(Box::into_raw(Box::new(0)))
        }
        pub fn set(&mut self, v: i32) {
            unsafe { *self.0 = v };
        }
        pub fn get(&self) -> i32 {
            unsafe { *self.0 }
        }
    }
}

enum Message {
    Set(i32),
    Get,
    Shutdown,
}

enum Reply {
    Set,
    Get(i32),
    Shutdown,
}

fn confinement_thread(
    receiver: std::sync::mpsc::Receiver<(Message, std::sync::mpsc::SyncSender<Reply>)>,
) {
    // Create the confined state
    let mut bar = lib::Bar::new();

    // Handle messages and forward them
    loop {
        let (mesg, reply_channel) = receiver.recv().unwrap();
        match mesg {
            Message::Set(v) => {
                eprintln!("    worker: setting value to {}", v);
                bar.set(v);
                reply_channel.send(Reply::Set).unwrap();
            }
            Message::Get => {
                let v = bar.get();
                eprintln!("    worker: getting value = {}", v);
                reply_channel.send(Reply::Get(v)).unwrap();
            }
            Message::Shutdown => {
                eprintln!("    worker: shutting down");
                reply_channel.send(Reply::Shutdown).unwrap();
                break;
            }
        }
    }
}

// This can be cloned happily
// and supports Send+Sync
struct GlobalProxy {
    channel: std::sync::mpsc::SyncSender<(Message, std::sync::mpsc::SyncSender<Reply>)>,
}

impl GlobalProxy {
    pub fn set(&self, v: i32) {
        eprintln!("  proxy: setting value to {}", v);
        let (a, b) = sync_channel(0);
        self.channel.send((Message::Set(v), a)).unwrap();
        let m = b.recv().unwrap();
        assert!(matches!(m, Reply::Set));
    }

    pub fn get(&self) -> i32 {
        eprintln!("  proxy: getting value");
        let (a, b) = sync_channel(0);
        self.channel.send((Message::Get, a)).unwrap();
        let m = b.recv().unwrap();
        if let Reply::Get(v) = m {
            eprintln!("  proxy: got value={}", v);
            v
        } else {
            unreachable!();
        }
    }

    pub fn die(&self) {
        eprintln!("Telling worker thread to shut down");
        let (a, b) = sync_channel(0);
        self.channel.send((Message::Shutdown, a)).unwrap();
        let m = b.recv().unwrap();
        assert!(matches!(m, Reply::Shutdown));
    }
}

lazy_static! {
    static ref G: GlobalProxy = {
        // Create com channels
        let (to_global, from_world) = sync_channel(0);
        // Keep one end for the proxy,
        let global = GlobalProxy{ channel: to_global};
        // The other goes to the worker thread
        std::thread::spawn(|| {confinement_thread(from_world)});
        global
    };
}

pub fn main() {
    eprintln!("global.get() = {}", G.get());
    eprintln!("global.set(10)",);
    G.set(10);
    eprintln!("global.get() = {}", G.get());

    G.die()
}

Working version

可能有很多机会使用宏来简化这个过程，但我发现这个版本更具指导意义。

另一个改进是将回复频道放入消息对象 - 这将允许我们删除 Reply 枚举。

在某些情况下，可以通过将函数传递给运行的限制线程而不是消息来删除 Message 对象。类似于：

impl GlobalProxy { 
  fn run_confined(&self f: dyn Fn(&lib::Bar) + Send + Sync)
   {...}
}

但是以一种很好的方式处理具有 return 值的函数是很棘手的。

Answer 3

我工作的示例不是来自全局状态，而是来自外部范围的变量。也许有用。

use std::collections::HashMap;

fn main() {
    let a = String::from("Hello world");
    let b = String::from("Another world");
    let mut keys: HashMap<String, String> = HashMap::new();

    let callback = |line: String| {
        keys.insert(line.to_string(), line.to_string());
        println!("{}", b);
        println!("{}", line);
        println!("{:?}", keys);
    };

    compute(a, callback)
}

fn compute<F>(a: String, mut f: F)
    where F: FnMut(String)
{
    f(a)
}

从 Rust 中的回调访问全局状态

Accessing global state from a callback in rust

hook

callback

ffi

rust