fix Encoder to be Send + Sync

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>

src/encoder/aio.rs

@@ -261,3 +261,23 @@ mod tokio_writer {
 
 #[cfg(feature = "tokio-io")]
 pub use tokio_writer::TokioWriter;
+
+#[test]
+/// Assert that `Encoder` is `Send`
+fn send_test() {
+    let test = async {
+        let mut encoder = Encoder::from_tokio(
+            tokio::fs::File::create("foo").await.unwrap(),
+            &Metadata::dir_builder(0o700).build(),
+        )
+        .await
+        .unwrap();
+        encoder
+            .create_directory("baba", &Metadata::dir_builder(0o700).build())
+            .await
+            .unwrap();
+    };
+
+    fn test_send<T: Send>(_: T) {}
+    test_send(test);
+}

src/encoder/mod.rs

@@ -2,13 +2,12 @@
 //!
 //! This is the implementation used by both the synchronous and async pxar wrappers.
 
-use std::cell::RefCell;
 use std::io;
 use std::mem::{forget, size_of, size_of_val, take};
 use std::os::unix::ffi::OsStrExt;
 use std::path::Path;
 use std::pin::Pin;
-use std::rc::Rc;
+use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
 
 use endian_trait::Endian;
@@ -121,9 +120,11 @@ where
     T: SeqWrite + ?Sized,
 {
     let data = data.to_le();
+    let buf =
+        unsafe { std::slice::from_raw_parts(&data as *const E as *const u8, size_of_val(&data)) };
     seq_write_all(
         output,
-        unsafe { std::slice::from_raw_parts(&data as *const E as *const u8, size_of_val(&data)) },
+        buf,
         position,
     )
     .await
@@ -177,10 +178,12 @@ where
     E: Endian,
 {
     let data = data.to_le();
+    let buf =
+        unsafe { std::slice::from_raw_parts(&data as *const E as *const u8, size_of_val(&data)) };
     seq_write_pxar_entry(
         output,
         htype,
-        unsafe { std::slice::from_raw_parts(&data as *const E as *const u8, size_of_val(&data)) },
+        buf,
         position,
     )
     .await
@@ -250,7 +253,7 @@ pub(crate) struct EncoderImpl<'a, T: SeqWrite + 'a> {
 
     /// Since only the "current" entry can be actively writing files, we share the file copy
     /// buffer.
-    file_copy_buffer: Rc<RefCell<Vec<u8>>>,
+    file_copy_buffer: Arc<Mutex<Vec<u8>>>,
 }
 
 impl<'a, T: SeqWrite + 'a> Drop for EncoderImpl<'a, T> {
@@ -278,7 +281,7 @@ impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
             state: EncoderState::default(),
             parent: None,
             finished: false,
-            file_copy_buffer: Rc::new(RefCell::new(crate::util::vec_new(1024 * 1024))),
+            file_copy_buffer: Arc::new(Mutex::new(crate::util::vec_new(1024 * 1024))),
         };
 
         this.encode_metadata(metadata).await?;
@@ -356,9 +359,9 @@ impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
         file_size: u64,
         content: &mut dyn SeqRead,
     ) -> io::Result<LinkOffset> {
-        let buf = Rc::clone(&self.file_copy_buffer);
+        let buf = Arc::clone(&self.file_copy_buffer);
         let mut file = self.create_file(metadata, file_name, file_size).await?;
-        let mut buf = buf.borrow_mut();
+        let mut buf = buf.lock().expect("failed to lock temporary buffer mutex");
         loop {
             let got = decoder::seq_read(&mut *content, &mut buf[..]).await?;
             if got == 0 {
@@ -543,7 +546,7 @@ impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
             },
             parent: Some(&mut self.state),
             finished: false,
-            file_copy_buffer: Rc::clone(&self.file_copy_buffer),
+            file_copy_buffer: Arc::clone(&self.file_copy_buffer),
         })
     }