问题[multithreading](coding)

我正在实现命令模式的实现，其中订阅者将从读取器线程接收数据，并且我在使用 Rust mocks 测试我的实现时遇到了问题。请考虑这个简化的示例：

use std::thread;

pub trait Subscriber: Sync + Send {
    fn callback(&self);
}

pub struct Publisher {
    thread_handler: Option<thread::JoinHandle<()>>,
}

impl Publisher {
    pub fn new() -> Self {
        Self {
            thread_handler: None,
        }
    }

    pub fn start(&mut self, sub: Box<dyn Subscriber>) {
        self.thread_handler = Some(thread::spawn(move || {
            sub.callback();
        }));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use mockall::*;

    mock! {
        TestSubscriber {}

        impl Subscriber for TestSubscriber {
            fn callback(&self);
        }
    }

    #[test]
    fn test_subscriber() {
        let mut publisher = Publisher::new();

        let mut mock_test_subscriber = Box::new(MockTestSubscriber::new());

        publisher.start(mock_test_subscriber);

        mock_test_subscriber.expect_callback().times(1);

        thread::sleep(std::time::Duration::from_millis(10));
    }
}

这会引发

error[E0382]: borrow of moved value: `mock_test_subscriber`
  --> src/foo.rs:46:9
   |
42 |         let mut mock_test_subscriber = Box::new(MockTestSubscriber::new());
   |             ------------------------ move occurs because `mock_test_subscriber` has type `Box<MockTestSubscriber>`, which does not implement the `Copy` trait
43 |
44 |         publisher.start(mock_test_subscriber);
   |                           -------------------- value moved here
45 |
46 |         mock_test_subscriber.expect_callback().times(1);
   |         ^^^^^^^^^^^^^^^^^^^^ value borrowed here after move

我真的不知道如何解决这个问题，我尝试过绕过它dyn Subscriber，Arc但仍然无法让它工作。任何建议都将不胜感激。或者也许有更好的设计来实现相同的结果。

在 Go 中我无法弄清楚如何获得以下设置。

• 第一级 goroutine 应该在终止之前停止其第二级 goroutine。
• 如果函数runGoroutine没有循环（我不希望它循环），那么当消息发送到通道时，goroutine 将如何退出？如果没有循环，那么case <-stopChan:只会在开始时被调用一次。
• 如果函数runGoroutine调用其他异步函数会怎么样？当返回时这些函数也会退出吗？

也许我的方法不对。在我的情况下，我有x用户可以启动的任务组，当用户启动任务组时，会创建一个 goroutine（第 1 级）以返回 CL 用户访问权限。此任务组（第 1 级）是yAPI 请求的集合（实际上 y 是 200-300）。对于每个 API 请求，我都会创建一个新的 go-routine（第 2 级）。我的目标是速度；如果我每秒执行数百个 API 请求，将每个请求放入其自己的 goroutine 中是否更快/更有效？或者在 1 个线程上执行是否相同？

                       Parent/Main process
                        |              |
          Child goroutine            Child goroutine
                |                           |
      X bottom goroutines          X bottom goroutines           (API level)
 

package main

import (
    "fmt"
    "time"
)

func runGoroutine(stopChan <-chan struct{}) {
    for {
        select {
        case <-stopChan:
            return // Exit the goroutine
        default:
            fmt.Println("Goroutine is working...")
            time.Sleep(1 * time.Second)
        }
    }
}
func main() {
    stopChan := make(chan struct{})
    // Start the goroutine
    go runGoroutine(stopChan)
    // User should be able to close the routines they want
    close(stopChan)
}

新代码

package main

import (
    "context"
    "log"
    "net/http"
    "sync"
    "time"
)

func main() {
    // Initialize context with cancellation
    ctx, cancel := context.WithCancel(context.Background())

    // Call the worker pool in a separate goroutine to allow main to return immediately

    // We can start any task group like this

    go startWorkerPool(ctx, 20, 100)
    
    // User has CL access on main thread
    // Main function returns, but `cancel` is accessible to allow later cancellation if needed
    log.Println("Worker pool started; you can cancel it by calling cancel()")
    waitForNine()
}

func waitForNine() {
    var input int
    for input != 9 {
        fmt.Print("Enter a number: ")
        fmt.Scan(&input)
    }
}

// startWorkerPool starts a fixed-size worker pool to handle `numRequests` requests.
func startWorkerPool(ctx context.Context, numWorkers, numRequests int) {
    var wg sync.WaitGroup
    work := make(chan string)

    // Launch the specified number of workers
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(ctx context.Context) {
            defer wg.Done()
            client := http.Client{}
            for url := range work {
                if req, err := http.NewRequestWithContext(ctx, "GET", url, nil); err != nil {
                    log.Println("failed to create new request:", err)
                } else if resp, err := client.Do(req); err != nil {
                    log.Println("failed to make request:", err)
                } else {
                    resp.Body.Close()
                }
            }
        }(ctx)
    }

    // Enqueue URLs for workers and close the channel when done
    go func() {
        for i := 0; i < numRequests; i++ {
            work <- "https://httpbin.org/get"
        }
        close(work)
    }()

    // Wait for all workers to finish
    wg.Wait()
    log.Println("All requests processed")
}

这就像一个线程池上下文。我有一个全局原子变量，它指示工作线程是否应该退出：

atomic<bool> exit;

此类工作线程按如下方式循环执行：

while (true) {
  std::unique_lock lock{mut};
  cv.wait(lock, [] {
    return !queue.empty() || exit.load();
  };

  if (exit.load()) {
    return;
  }

  // take task from queue
}

还有其他有效的方法（具有更好的性能、更低的延迟）来检查我们是否应该退出工作线程吗？

据我了解：

• 活锁：一个线程无法取得进展，但它会永远执行（该线程无限期地处于活动状态）：例如，它处于一个循环中，无限地尝试访问永远不会再可用的共享资源。
• 死锁：线程没有进展，但甚至没有执行（线程无限期休眠，因此已死亡）：它正在等待操作系统根据永远不会发生的条件唤醒它。
• 饥饿：一个线程没有取得进展并且被阻塞（类似于死锁），但是等待被唤醒的条件确实发生了，只是碰巧有其他线程总是在我之前取代我的位置。

首先，这些定义是否正确？如果正确，它们是否也适用于单线程程序？例如：

// begin of useful work
until (some_buggy_condition_that_will_never_be_satisfied())
    do_something();
// end of useful work

在上面的例子中，循环将永远执行，因为与共享资源无关的条件有一个错误，并且永远不会满足。因此线程：

• 正在执行。
• 为了做有用的工作，循环必须结束
• 循环永远不会结束，因为结束条件永远不会满足

因此，该列表符合我上面的活锁定义，只是我在并发上下文之外使用了该术语。这样可以吗？

我有一些 url，然后我将其分成块并同时迭代它们，使用克隆将结果保存在 vec 中：

pub fn collect_parm_list(url_list: Vec<String>) -> Result<Vec<String>, Box<dyn std::error::Error>> {
let shared_id_list = Arc::new(Mutex::new(vec![]));
for chunk in url_list.chunks(20) {
    let mut handles = vec![];
    for url in chunk {
        let id_list = Arc::clone(&shared_id_list);
        let handle = thread::spawn(move || {
            // total: optionSale -> result -> total
            let mut id_list_v = id_list.lock().unwrap();
            id_list_v.push(url.to_string())
        });

        handles.push(handle);
    }
    // Wait for all threads to complete
    for handle in handles {
        handle.join().unwrap();
    }
}

let x = shared_id_list.lock().unwrap();
Ok(x.to_vec())
}

我在借用数据时遇到错误：

url_list does not live long enough borrowed value does not live long enough

在：对于url_list中的块。chunks（20）{

我有大约 75000 个文件，我需要在每个文件中搜索一组存储在数组中的关键短语。我的 Intel i9 能够运行 20 个线程。我试图通过将每个文件放入字符串并同时匹配每个关键短语来加快整个过程。我想知道如何使用 hyper/race 来进一步加快这一过程。或者连接会自动并发地跨线程分配任务吗？

[1] > my $a = (1..10).join
12345678910
[3] > my @b = (3, /5./, /8\d/)
[3 /5./ /8\d/]
[4] > say $a.match( @b.all )
all(3, 56, 89)

[4] > say hyper $a.match( @b.all )
No such method 'hyper' for invocant of type 'Match'.    # what to do?

我正在尝试通过制作一个简单的向量加法函数来学习 Rust 的 Rayon 库。我当前的代码是这样的，假设a、b和c是初始化为相同长度的向量，c是可变的并且num_threads是一个usize变量：

let pool = ThreadPoolBuilder::new().num_threads(num_threads).build()
           .expect("Could not create thread pool");
pool.install(|| {
    (0..c.len()).into_par_iter().for_each(|x| {
        c[x] = a[x] + b[x];
    });
});

但我得到了错误

error[E0596]: cannot borrow c as mutable, as it is a captured variable in a Fn closure
    c[x] = a[x].wrap_add(b[x]);
    ^ cannot borrow as mutable

我还想指出，最终我的目标是制作基准软件，这就是为什么我使用线程池来指定线程数，但我认为额外的闭包pool.install()使用是问题根源的一部分。修改全局线程池也不是一个选项，因为这只能完成一次，而且我想使用不同的线程数重新运行基准测试。我还想避免使用范围 - 如果这是唯一的解决方案，那就这样吧 - 因为这会增加性能损失。

我从根本上理解 Rayon 在这里不喜欢什么：Rust Book 第 4.2 章说你不能对一个变量有多个可变引用，这本质上是每个线程都会得到的。然而，除了如何让这段代码工作的特定问题之外，这还引发了一些其他问题。

我似乎甚至无法将引用移至c线程池闭包中。为什么这是一个限制？当然，多线程的强大之处在于让多个线程同时对相关数据进行一些工作，那么为什么我不能以这种方式将数据传递给线程呢？

假设我可以获得对线程池的引用c，是否有某种方法可以实现，例如，线程 0 gets &mut c[0]、线程 1 gets&mut c[1]等等？如果 Rayon 的目的是抽象 Rust 基本多线程库的一些样板，那么 Rayon 不应该尝试让这变得更简单吗？

我看到的其他一些答案暗示迭代向量本身的内容会有所帮助，但由于我需要所有三个向量，所以我需要使用izip。这样做（替换(0..c.len().into_par_iter()...为izip!(&a.mat, &b.mat, &mut c.mat).into_par_iter()...）给了我错误

error[E0599]: the method into_par_iter exists for struct Map<Zip<Zip<Iter<'_, T>, Iter<'_, T>>, IterMut<'_, T>>, {[email protected]:303:9}>, but its trait bounds were not satisfied

izip!(&a.mat, &b.mat, &mut c.mat).into_par_iter().for_each(|x| {
   |                                                   ^^^^^^^^^^^^^
   |
   = note: the following trait bounds were not satisfied:
           `std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::ParallelIterator`
           which is required by `std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::IntoParallelIterator`
           `&std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::ParallelIterator`
           which is required by `&std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::IntoParallelIterator`
           `&mut std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::ParallelIterator`
           which is required by `&mut std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::IntoParallelIterator`

这似乎意味着引用现在正在被移至关闭中。是什么赋予了？之前为什么不搬走呢？

在actix-web路由功能中，我发现从互斥锁保护对象获取独立数据后，互斥锁必须处于锁定状态才能使用这些数据。如果互斥体仅在获取数据时加锁，则使用数据时会出现编译错误。下面是可以直接编译的代码：

use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::Mutex;
use async_trait::async_trait;
use actix_web::{web, App, HttpServer, HttpResponse, Result as ActixResult};

#[async_trait]
pub trait Reader: Send + Debug { }

#[derive(Clone, Debug)]
pub struct RemoteReader;

#[async_trait]
impl Reader for RemoteReader { }

#[derive(Default)]
pub struct DataReaders(HashMap<String, Box<dyn Reader>>);

impl DataReaders {
    pub fn get_readers(&self, ids: &str) -> ActixResult<Vec<&dyn Reader>> {
        let mut readers = Vec::new();
        for id in ids.split(',') {
            readers.push(self.0.get(id).unwrap().as_ref());
        }
        Ok(readers)
    }
}

async fn get(data: web::Data<Mutex<DataReaders>>) -> ActixResult<HttpResponse> {
    let mut readers = data.lock().unwrap().get_readers("1,2")?;
    if readers.is_empty() {
        return Ok(HttpResponse::NotFound().finish())
    }
    // Below will call multiple asynchronous functions, so I do not want to
    // hold the lock of DataReaders for a long time.
    readers.clear();
    Ok(HttpResponse::Ok().finish())
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let mut readers = DataReaders{ 0: Default::default() };
    readers.0.insert("1".to_string(), Box::new(RemoteReader{}));
    readers.0.insert("2".to_string(), Box::new(RemoteReader{}));
    let counter = web::Data::new(Mutex::new(readers));

    HttpServer::new(move || {
        App::new()
            .app_data(counter.clone())
            .route("/", web::get().to(get))
    }).bind(("127.0.0.1", 8080))?.run().await
}

编译这段代码并得到错误：

error[E0716]: temporary value dropped while borrowed
  --> src/main.rs:30:23
   |
30 |     let mut readers = data.lock().unwrap().get_readers("1,2")?;
   |                       ^^^^^^^^^^^^^^^^^^^^                    - temporary value is freed at the end of this statement
   |                       |
   |                       creates a temporary value which is freed while still in use
31 |     if readers.is_empty() {
   |        ------- borrow later used here
   |
   = note: consider using a `let` binding to create a longer lived value

我只想使用互斥锁来保护 HashMap，因为它会在其他路由函数中被更改。我认为从“get_readers”返回的 Vec<&dyn Reader> 对象不应受到保护，因为其中的所有项目都是只读的。那么我该如何解决这个问题呢？谢谢！

我正在阅读《英特尔® 64 和 IA-32 架构软件开发人员手册：系统编程指南》以了解有关操作系统如何工作的更多信息，但有一些事情我无法弄清楚。

因此，我了解到操作系统内核代码映射到所有进程的虚拟地址空间，并以系统调用或中断/异常的特权运行，内核可以通过 IPI 进行通信，进程的线程作为 x86 架构中的任务运行，线程切换需要更改任务寄存器以指向下一个任务状态段等。我也知道分页、保护和所有这些是如何工作的。

然而，当存在多核时，我开始对执行某些“全局”内核操作（例如线程切换）的核心感到困惑。我在大文档中找不到它。我主要关心的是线程切换。当线程量被消耗时，哪个CORE执行线程调度，是BSP处理器在执行还是这是由BSP和所有AP执行的分布式任务？

当收到切换线程所需的时间中断时，是否所有核心都收到？在这种情况下他们如何管理共享资源？（因此，如果其中一个核心调度了一个线程，其他核心也不会调度它）

提前致谢！！

单击 myButton 后，运行 2 个线程。首先更改按钮图像，第二个 - 循环。我希望我的程序首先更改图像并在运行循环之后。但这不行，他们一瞬间就完蛋了。

myButton.setOnMouseClicked(event -> {
            if (event.getButton() == MouseButton.PRIMARY) {
                Thread thread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        ImageView imageView = new ImageView(image);
                        imageView.setFitHeight(my_button_height - 16);
                        imageView.setFitWidth(my_button_width - 16);
                        myButton.setStyle("-fx-background-color: #ffffff00; ");
                        myButton.setGraphic(imageView);
                    }
                });
                thread.run();
                try {
                    thread.join();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                Thread thread1 = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        for (int i = 0; i < 15; i++){
                            System.out.println(i);
                            try {
                                Thread.sleep(500);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        }
                    }
                });
                thread1.run();
            }
        });
    }

我尝试添加.join。这没有帮助