Drain a Vec<_> from within a tokio task

user_8053020 0
Viewed 40

I have a buffer of type Vec<i32> which I want to write to disk and .drain() it after a certain time. The problem I'm facing is that of ownership. How can I write the contents of the buffer to disk and .drain() it without taking ownership or is it not possible in this case because of the tokio tasks?

use tokio::sync::mpsc;
use std::time::Duration;
use std::fs::File;
use std::io::BufWriter;
use serde_json;

#[tokio::main]
async fn main() {
    let mut buffer: Vec<i32> = Vec::new();

    let mut interval = tokio::time::interval(Duration::from_secs(60));

    let (tx, mut rx) = mpsc::unbounded_channel();

    let sender_handle = tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).unwrap();
        }
    });

    let receiver_handle = tokio::spawn(async move {
        while let Some(x) = rx.recv().await {
            buffer.push(x);
        }
    });

    let interval_handle = tokio::spawn(async move {
        loop {
            interval.tick().await;
            
            let file: File::create("buffer.txt");
            let mut writer = BufWriter::new(file.unwrap());
            serde_json::to_writer(&mut writer, &buffer);

            buffer.drain(..);
        }
    });

    let (_, _, _) = tokio::join!(sender_handle, receiver_handle, interval_handle);
}
rust
1 Answers

You need to add some sync mechanism so the access to the vector is controlled. An Arc<Mutex<Vec<_>>> would do:

use futures::lock::Mutex;
use std::sync::Arc;
use tokio::sync::mpsc;
use std::time::Duration;
use std::fs::File;
use std::io::BufWriter;
use serde_json;

#[tokio::main]
async fn main() {
    let buffer: Arc<Mutex<Vec<i32>>> = Arc::new(Mutex::new(Vec::new()));

    let mut interval = tokio::time::interval(Duration::from_secs(60));

    let (tx, mut rx) = mpsc::unbounded_channel();

    let sender_handle = tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).unwrap();
        }
    });
    let receiver_buff = buffer.clone();
    let receiver_handle = tokio::spawn(async move {
        while let Some(x) = rx.recv().await {
            let mut buff = receiver_buff.lock().await;
            buff.push(x);
        }
    });

    let interval_handle = tokio::spawn(async move {
        let buffer = buffer.clone();
        loop {
            interval.tick().await;
            
            let file =  File::create("buffer.txt");
            let mut writer = BufWriter::new(file.unwrap());
            let mut buff = buffer.lock().await;
            serde_json::to_writer(&mut writer, &*buff).expect("Serialization to go well");

            buff.drain(..);
        }
    });

    let (_, _, _) = tokio::join!(sender_handle, receiver_handle, interval_handle);
}

Playground

user_16951720
Related