autocommit 2022-02-24 16:30:11 CET

src/utils/bitflags.rs

@@ -1,11 +1,16 @@
 pub trait BitflagsExt {
     fn contains(self, other: Self) -> bool;
+    fn intersects(self, other: Self) -> bool;
 }
 
 macro_rules! num {
     ($ty:ident) => {
         impl BitflagsExt for $ty {
             fn contains(self, other: Self) -> bool {
+                self & other == other
+            }
+
+            fn intersects(self, other: Self) -> bool {
                 self & other != 0
             }
         }

src/utils/buffd/buf_in.rs

@@ -74,7 +74,11 @@ impl BufFdIn {
             name: uapi::sockaddr_none_mut(),
             flags: 0,
         };
-        let (iov, _, mut cmsg) = match uapi::recvmsg(self.fd.raw(), &mut hdr, c::MSG_DONTWAIT) {
+        let (iov, _, mut cmsg) = match uapi::recvmsg(
+            self.fd.raw(),
+            &mut hdr,
+            c::MSG_DONTWAIT | c::MSG_CMSG_CLOEXEC,
+        ) {
             Ok((iov, _, _)) if iov.is_empty() => return Err(BufFdError::Closed),
             Ok(v) => v,
             Err(Errno(c::EAGAIN)) => return Ok(true),

src/utils/buffd/buf_out.rs

@@ -157,7 +157,11 @@ impl BufFdOut {
         Ok(false)
     }
 
-    pub async fn flush2(&mut self, buf: &[u8], fds: &mut Vec<Rc<OwnedFd>>) -> Result<(), BufFdError> {
+    pub async fn flush2(
+        &mut self,
+        buf: &[u8],
+        fds: &mut Vec<Rc<OwnedFd>>,
+    ) -> Result<(), BufFdError> {
         let mut read_pos = 0;
         while read_pos < buf.len() {
             if self.flush_sync2(&mut read_pos, buf, fds)? {

src/utils/mod.rs

@@ -12,4 +12,5 @@ pub mod ptr_ext;
 pub mod queue;
 pub mod smallmap;
 pub mod stack;
+pub mod tri;
 pub mod vec_ext;

src/utils/queue.rs

@@ -39,6 +39,10 @@ impl<T> AsyncQueue<T> {
         AsyncQueuePop { queue: self }
     }
 
+    pub fn non_empty<'a>(&'a self) -> AsyncQueueNonEmpty<'a, T> {
+        AsyncQueueNonEmpty { queue: self }
+    }
+
     pub fn clear(&self) {
         mem::take(&mut *self.data.borrow_mut());
         self.waiter.take();
@@ -61,3 +65,20 @@ impl<'a, T> Future for AsyncQueuePop<'a, T> {
         }
     }
 }
+
+pub struct AsyncQueueNonEmpty<'a, T> {
+    queue: &'a AsyncQueue<T>,
+}
+
+impl<'a, T> Future for AsyncQueueNonEmpty<'a, T> {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.queue.data.borrow_mut().len() > 0 {
+            Poll::Ready(())
+        } else {
+            self.queue.waiter.set(Some(cx.waker().clone()));
+            Poll::Pending
+        }
+    }
+}

src/utils/tri.rs

@@ -0,0 +1,49 @@
+use std::future::Future;
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pub trait Try: Sized {
+    fn tri<F>(f: F) -> Result<(), Self>
+    where
+        F: FnOnce() -> Result<(), Self>;
+
+    fn tria<F>(f: F) -> Tria<Self, F>
+    where
+        F: Future<Output = Result<(), Self>>;
+}
+
+impl<E> Try for E {
+    fn tri<F>(f: F) -> Result<(), Self>
+    where
+        F: FnOnce() -> Result<(), Self>,
+    {
+        f()
+    }
+
+    fn tria<F>(f: F) -> Tria<E, F>
+    where
+        F: Future<Output = Result<(), Self>>,
+    {
+        Tria {
+            f,
+            _phantom: Default::default(),
+        }
+    }
+}
+
+pub struct Tria<E, F> {
+    f: F,
+    _phantom: PhantomData<E>,
+}
+
+impl<E, F> Future for Tria<E, F>
+where
+    F: Future<Output = Result<(), E>>,
+{
+    type Output = Result<(), E>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        unsafe { Pin::new_unchecked(&mut Pin::get_unchecked_mut(self).f).poll(cx) }
+    }
+}