Block loop with tokio mutex lock, but use locked value inside a spawn task

I've got this code:

  loop {
            let receiver = self.receiver.clone();
      
            // let mut lock = receiver.lock().await; //i want this to block till receiver is free to access. But this doesn't work since receiver doesn't live long enough.
  
            tokio::spawn(async move {
                let mut lock = receiver.lock().await; //Instead of here where many threads are created trying to access it.
                let recv = lock.recv();
                let guard = recv.await.unwrap();

                let task: I = serde_json::from_slice(&guard).unwrap();
             
            });
        }

If my understanding is correct this code would continuously spawn threads, all of which would be waiting to lock the mutex. This doesn't seem very appropriate and would lead to huge contention.

instead i would like to first acquire a lock then spawn a thread. Does this even make sense ? Wouldn't that mean if the code compiles each thread would have unrestricted access to mutate the data which could cause race conditions ?

I also tried using a stream:

let receiver = self.receiver.clone();

        let guard_stream = stream! {
            loop {
                 let mut lock = receiver.clone().lock().await;
                yield lock.recv().await.unwrap();
            }
        };

        pin_mut!(guard_stream);

        while let Some(guard) = guard_stream.next().await {
           
            tokio::spawn(async move {            
                let task: I = serde_json::from_slice(&guard).unwrap();
            });
        }

This doesn't work since lock doesn't live longer it's yielded value.

Use lock_owned. This will let you move the mutex guard to another task.

2 Likes