Запустите вечный поток с циклом просроченных объектов, удалив эти объекты в ржавчине

Я пишу службу и хочу вечно запускать цикл, который проверяет наличие некоторых просроченных объектов и удаляет их, если они слишком старые.

pub struct Obj {
    expired: NaiveDateTime,
}

pub struct Maintainer {
    objs: HashMap<id, Obj>,
}

pub trait Miller {
    fn new() -> Self;
}

impl Miller for Maintainer {
    fn new() -> Self {
        let i = Self {
            obj: Hashmap::new(),
        };
        i.start_exp_observer();
        i
    }
}

impl Maintainer {
    fn start_exp_observer(&self) {
        let observer = thread::spawn(move || loop {
            thread.sleep(sleep_time);
            self.objs
                .retain(|_, o| o.expired.gt(Utc::now().naive_utc()));
        });
        // does this even work here
        observer.join().unwrap();
    }
}

В Rust это не работает, потому что я использую неизменяемый self в start_exp_observer, созданный new().

Я попробовал fn start_exp_observer(&mut self), но это доставило мне неприятности, поскольку Self, созданный в new, не подлежит изменению. И если я определю это Self как изменчивое, то у меня возникнут проблемы с этой чертой.

И если мне кажется, что чем больше я пытаюсь, тем больше у меня проблем (например, если бы я клонировал Self перед возвратом объекта, то, наверное, я не смотрю на одни и те же объекты в потоке).

Как это можно сделать?


50
1

Ответ:

Решено

Вам необходимо убедиться, что тип self может передаваться между потоками, и ограничить только один доступ одновременно, то есть Mutex и Arc.

use chrono::*;
use std::collections::HashMap;

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

pub struct Obj {
    expired: NaiveDateTime,
}

#[derive(Clone)]
pub struct Maintainer {
    objs: Arc<Mutex<HashMap<u64, Obj>>>,
    //    ^   ^     ^
    //    |   |     |
    //    |   |     L The actual data 
    //    |   L To make sure only one access at a time
    //    L to make the data be able to pass between thread

}

pub trait Miller {
    fn new() -> Self;
}

impl Miller for Maintainer {
    fn new() -> Self {
        let objs = Default::default();
        let i = Self { objs };
        i.start_exp_observer();
        i
    }
}

impl Maintainer {
    fn start_exp_observer(&self) {
        // Clone a self, the clone will be sent to another thread
        let maintainer = self.clone();
        // Spawn a new thread, the `move` will move the `maintainer` variable to another thread
        // while keeping `self` in the current thread
        std::thread::spawn(move || loop {
            std::thread::sleep(std::time::Duration::from_millis(1000));
            maintainer
                .objs 
                .lock()
                // ^^^^ we need to lock the mutex, to make sure we do not have another thread
                //      i.e. the main thread, accessing it at the same time
                .unwrap()
                .retain(|_, o| o.expired.gt(&Utc::now().naive_utc()));
        });
        // Do not `join()` the handle
        // join mean await for it to finish 
        // because it is a infinite loop, the main thread will no progress
    }
}