Started converting LocalFs to use async locks.

crates/btfproto/src/server.rs

@@ -7,7 +7,28 @@ use crate::{
 use btlib::{crypto::Creds, BlockPath, Result};
 use btmsg::{receiver, MsgCallback, MsgReceived, Receiver};
 use core::future::Future;
-use std::{io::Read, net::IpAddr, sync::Arc};
+use std::{io::Read, net::IpAddr, sync::Arc, ops::Deref};
+
+pub trait ReadingGuard<'a> {
+    type ReadGuard<'b>: ReadGuard<'b> where Self: 'b;
+    type ReadFut<'b>: Send + Future<Output = Result<Self::ReadGuard<'b>>> where Self: 'b;
+    fn read(&self) -> Self::ReadFut<'_>;
+
+    type WriteGuard<'b>: WriteGuard<'b> where Self: 'b;
+    type WriteFut<'b>: Send + Future<Output = Result<Self::WriteGuard<'b>>> where Self: 'b;
+    fn write(&self) -> Self::WriteFut<'_>;
+}
+
+pub trait ReadGuard<'a> {
+    type BufGuard<'b>: 'b + Deref<Target = [u8]> where Self: 'b;
+    type BufFut<'b>: Send + Future<Output = Result<Option<Self::BufGuard<'b>>>> where Self: 'b;
+    fn get_buf(&self) -> Self::BufFut<'_>;
+}
+
+pub trait WriteGuard<'a> {
+    type SeekFut: Send + Future<Output = Result<()>>;
+    fn seek(&'a mut self) -> Self::SeekFut;
+}
 
 pub trait FsProvider: Send + Sync {
     type LookupFut<'c>: Send + Future<Output = Result<LookupReply>>
@@ -25,14 +46,19 @@ pub trait FsProvider: Send + Sync {
         Self: 'c;
     fn open<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Open) -> Self::OpenFut<'c>;
 
+    type ReadingGuard<'a>: 'a + ReadingGuard<'a> where Self: 'a;
+    //type ReadFut<'c>: Send + Future<Output = Result<Self::ReadingGuard<'c>>> where Self: 'c;
+    //fn read2<'c>(&'c self, from: &'c Arc<BlockPath>, msg: ReadMsg) -> Self::ReadFut<'c>;
+
     fn read<'c, R, F>(&'c self, from: &'c Arc<BlockPath>, msg: ReadMsg, callback: F) -> Result<R>
     where
         F: 'c + FnOnce(&[u8]) -> R;
 
-    type WriteFut<'c>: Send + Future<Output = Result<WriteReply>>
+    type WriteFut<'r, R>: Send + Future<Output = Result<WriteReply>>
     where
-        Self: 'c;
-    fn write<'c, R>(
+        Self: 'r,
+        R: 'r + Send + Read;
+    fn write<'c, 'r, R>(
         &'c self,
         from: &'c Arc<BlockPath>,
         inode: Inode,
@@ -40,9 +66,10 @@ pub trait FsProvider: Send + Sync {
         offset: u64,
         size: u64,
         reader: R,
-    ) -> Self::WriteFut<'c>
+    ) -> Self::WriteFut<'r, R>
     where
-        R: 'c + Read;
+        'c: 'r,
+        R: 'r + Read + Send;
 
     type FlushFut<'c>: Send + Future<Output = Result<()>>
     where
@@ -57,12 +84,12 @@ pub trait FsProvider: Send + Sync {
     type LinkFut<'c>: Send + Future<Output = Result<LinkReply>>
     where
         Self: 'c;
-    fn link<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Link) -> Self::LinkFut<'c>;
+    fn link<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Link<'c>) -> Self::LinkFut<'c>;
 
     type UnlinkFut<'c>: Send + Future<Output = Result<()>>
     where
         Self: 'c;
-    fn unlink<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Unlink) -> Self::UnlinkFut<'c>;
+    fn unlink<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Unlink<'c>) -> Self::UnlinkFut<'c>;
 
     type ReadMetaFut<'c>: Send + Future<Output = Result<ReadMetaReply>>
     where
@@ -135,6 +162,9 @@ impl<P: FsProvider> FsProvider for &P {
         (*self).open(from, msg)
     }
 
+    type ReadingGuard<'c> = P::ReadingGuard<'c> where Self: 'c;
+    //type ReadFut<'c> = P::ReadFut<'c> where Self: 'c;
+
     fn read<'c, R, F>(&'c self, from: &'c Arc<BlockPath>, msg: ReadMsg, callback: F) -> Result<R>
     where
         F: 'c + FnOnce(&[u8]) -> R,
@@ -142,8 +172,8 @@ impl<P: FsProvider> FsProvider for &P {
         (*self).read(from, msg, callback)
     }
 
-    type WriteFut<'c> = P::WriteFut<'c> where Self: 'c;
-    fn write<'c, R>(
+    type WriteFut<'r, R> = P::WriteFut<'r, R> where Self: 'r, R: 'r + Send + Read;
+    fn write<'c, 'r, R>(
         &'c self,
         from: &'c Arc<BlockPath>,
         inode: Inode,
@@ -151,9 +181,10 @@ impl<P: FsProvider> FsProvider for &P {
         offset: u64,
         size: u64,
         reader: R,
-    ) -> Self::WriteFut<'c>
+    ) -> Self::WriteFut<'r, R>
     where
-        R: 'c + Read,
+        'c: 'r,
+        R: 'r + Read + Send,
     {
         (*self).write(from, inode, handle, offset, size, reader)
     }
@@ -169,12 +200,12 @@ impl<P: FsProvider> FsProvider for &P {
     }
 
     type LinkFut<'c> = P::LinkFut<'c> where Self: 'c;
-    fn link<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Link) -> Self::LinkFut<'c> {
+    fn link<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Link<'c>) -> Self::LinkFut<'c> {
         (*self).link(from, msg)
     }
 
     type UnlinkFut<'c> = P::UnlinkFut<'c> where Self: 'c;
-    fn unlink<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Unlink) -> Self::UnlinkFut<'c> {
+    fn unlink<'c>(&'c self, from: &'c Arc<BlockPath>, msg: Unlink<'c>) -> Self::UnlinkFut<'c> {
         (*self).unlink(from, msg)
     }
 
@@ -285,9 +316,10 @@ impl<P: 'static + Send + Sync + FsProvider> MsgCallback for ServerCallback<P> {
                     offset,
                     data,
                 }) => FsReply::Write(
-                    provider
-                        .write(&from, inode, handle, offset, data.len() as u64, data)
-                        .await?,
+                    todo!()
+                    //provider
+                    //    .write(&from, inode, handle, offset, data.len() as u64, data)
+                    //    .await?,
                 ),
                 FsMsg::Flush(flush) => FsReply::Ack(provider.flush(&from, flush).await?),
                 FsMsg::ReadDir(read_dir) => {

crates/btfuse/src/fuse_fs.rs

@@ -305,10 +305,11 @@ mod private {
             block_on(async move {
                 let path = self.path_from_luid(ctx.uid);
                 let size: usize = size.try_into().display_err()?;
-                let WriteReply { written, .. } = self
-                    .provider
-                    .write(path, inode, handle, offset, size as u64, r)
-                    .await?;
+                let written: u64 = todo!();
+                //let WriteReply { written, .. } = self
+                //    .provider
+                //    .write(path, inode, handle, offset, size as u64, r)
+                //    .await?;
                 Ok(written.try_into().display_err()?)
             })
         }

crates/btlib/src/drop_trigger.rs

@@ -0,0 +1,66 @@
+pub struct DropTrigger<F: FnOnce()> {
+    trigger: Option<F>,
+}
+
+impl<F: FnOnce()> DropTrigger<F> {
+    pub fn new(trigger: F) -> Self {
+        Self { trigger: Some(trigger) }
+    }
+
+    pub fn disarm(&mut self) -> bool {
+        if self.trigger.take().is_some() {
+            true
+        } else {
+            false
+        }
+    }
+}
+
+impl<F: FnOnce()> Drop for DropTrigger<F> {
+    fn drop(&mut self) {
+        if let Some(trigger) = self.trigger.take() {
+            trigger()
+        }
+    }
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn trigger_called_on_drop() {
+        let mut x = 0;
+        let dt = DropTrigger::new(|| x += 1);
+
+        drop(dt);
+
+        assert_eq!(1, x);
+    }
+
+    #[test]
+    fn trigger_not_called_when_disarmed() {
+        let mut x = 0;
+        let mut dt = DropTrigger::new(|| x += 1);
+
+        let was_armed = dt.disarm();
+        drop(dt);
+
+        assert!(was_armed);
+        assert_eq!(0, x);
+    }
+
+    #[test]
+    fn disarm_returns_false_if_was_not_armed() {
+        let mut x = 0;
+        let mut dt = DropTrigger::new(|| x += 1);
+
+        let first = dt.disarm();
+        let second = dt.disarm();
+        drop(dt);
+
+        assert!(first);
+        assert!(!second);
+    }
+}

crates/btlib/src/lib.rs

@@ -12,6 +12,7 @@ pub mod sectored_buf;
 #[cfg(test)]
 mod test_helpers;
 mod trailered;
+pub mod drop_trigger;
 
 #[macro_use]
 extern crate static_assertions;