fix issues in experimental/mrpc

experimental/mrpc/plugin/load_balancer/src/engine.rs

@@ -233,7 +233,7 @@ impl Engine for LoadBalancerEngine {
     fn handle_request(
         &mut self,
         request: Vec<u8>,
-        _cred: std::os::unix::ucred::UCred,
+        _cred: std::os::unix::net::UCred,
     ) -> Result<()> {
         let request: control_plane::Request = bincode::deserialize(&request[..])?;
 

experimental/mrpc/plugin/policy/hello-acl-receiver/src/engine.rs

@@ -1,5 +1,5 @@
 //! This engine can only be placed at the sender side for now.
-use std::os::unix::ucred::UCred;
+use std::os::unix::net::UCred;
 use std::pin::Pin;
 use std::ptr::Unique;
 

experimental/mrpc/plugin/policy/hello-acl-sender/src/engine.rs

@@ -1,6 +1,6 @@
 //! This engine can only be placed at the sender side for now.
 use std::num::NonZeroU32;
-use std::os::unix::ucred::UCred;
+use std::os::unix::net::UCred;
 use std::pin::Pin;
 use std::ptr::Unique;
 

experimental/mrpc/plugin/policy/hotel-acl/src/engine.rs

@@ -1,6 +1,6 @@
 //! This engine can only be placed at the sender side for now.
 use std::num::NonZeroU32;
-use std::os::unix::ucred::UCred;
+use std::os::unix::net::UCred;
 use std::pin::Pin;
 use std::ptr::Unique;
 

experimental/mrpc/plugin/policy/logging/src/engine.rs

@@ -1,7 +1,7 @@
 use anyhow::{anyhow, Result};
 use futures::future::BoxFuture;
 use std::io::Write;
-use std::os::unix::ucred::UCred;
+use std::os::unix::net::UCred;
 use std::pin::Pin;
 
 use phoenix_api_policy_logging::control_plane;

experimental/mrpc/plugin/policy/null/src/engine.rs

@@ -1,4 +1,4 @@
-use std::os::unix::ucred::UCred;
+use std::os::unix::net::UCred;
 use std::pin::Pin;
 
 use anyhow::{anyhow, Result};

experimental/mrpc/plugin/policy/qos/src/engine.rs

@@ -1,7 +1,7 @@
 use std::cell::RefCell;
 use std::cmp::Reverse;
 use std::collections::binary_heap::BinaryHeap;
-use std::os::unix::ucred::UCred;
+use std::os::unix::net::UCred;
 use std::pin::Pin;
 use std::time::Duration;
 
@@ -103,7 +103,7 @@ impl Engine for QosEngine {
         BUFFER.with_borrow_mut(|buf| {
             let mut messages = buf.drain().collect::<Vec<_>>();
             let client_pid = self.client_pid;
-            let drained = messages.drain_filter(|msg| msg.0.source == client_pid);
+            let drained = messages.extract_if(|msg| msg.0.source == client_pid);
             for msg in drained {
                 self.tx_outputs()[0].send(msg.0.message)?;
             }

experimental/mrpc/plugin/policy/qos/src/lib.rs

@@ -1,6 +1,6 @@
 #![feature(peer_credentials_unix_socket)]
 #![feature(local_key_cell_methods)]
-#![feature(drain_filter)]
+#![feature(extract_if)]
 
 use thiserror::Error;
 

experimental/mrpc/plugin/policy/ratelimit/src/engine.rs

@@ -1,5 +1,5 @@
 use std::collections::VecDeque;
-use std::os::unix::ucred::UCred;
+use std::os::unix::net::UCred;
 use std::pin::Pin;
 
 use anyhow::{anyhow, Result};

experimental/mrpc/plugin/rpc_adapter/src/engine.rs

@@ -291,7 +291,7 @@ impl Engine for RpcAdapterEngine {
     fn handle_request(
         &mut self,
         request: Vec<u8>,
-        _cred: std::os::unix::ucred::UCred,
+        _cred: std::os::unix::net::UCred,
     ) -> Result<()> {
         let request: control_plane::Request = bincode::deserialize(&request[..])?;
 
@@ -464,7 +464,7 @@ impl RpcAdapterEngine {
                 .push_back(ReqContext { call_id, sg_len: 1 });
         }
 
-        let off = meta_buf_ptr.0.as_ptr().expose_addr();
+        let off = meta_buf_ptr.0.as_ptr().addr();
         let meta_buf = unsafe { meta_buf_ptr.0.as_mut() };
 
         // TODO(cjr): impl Serialize for SgList
@@ -538,7 +538,7 @@ impl RpcAdapterEngine {
         let ctx = self.rpc_ctx.insert(RpcId::new(cmid.as_handle(), call_id));
 
         let meta_sge = SgE {
-            ptr: (meta_ref as *const MessageMeta).expose_addr(),
+            ptr: (meta_ref as *const MessageMeta).addr(),
             len: mem::size_of::<MessageMeta>(),
         };
 
@@ -665,7 +665,7 @@ impl RpcAdapterEngine {
         let (_prefix, lens, _suffix): (_, &[u32], _) = unsafe { meta_buf.lens_buffer().align_to() };
         debug_assert!(_prefix.is_empty() && _suffix.is_empty());
 
-        let value_buf_base = meta_buf.value_buffer().as_ptr().expose_addr();
+        let value_buf_base = meta_buf.value_buffer().as_ptr().addr();
         let mut value_offset = 0;
 
         #[allow(clippy::needless_range_loop)]
@@ -1047,7 +1047,7 @@ impl RpcAdapterEngine {
             cmd::Command::NewMappedAddrs(conn_handle, app_vaddrs) => {
                 for (mr_handle, app_vaddr) in app_vaddrs.iter() {
                     let region = self.state.resource().recv_buffer_pool.find(mr_handle)?;
-                    let mr_local_addr = region.as_ptr().expose_addr();
+                    let mr_local_addr = region.as_ptr().addr();
                     let mr_remote_mapped = mrpc_marshal::ShmRecvMr {
                         ptr: *app_vaddr,
                         len: region.len(),

experimental/mrpc/plugin/rpc_adapter/src/pool.rs

@@ -36,7 +36,7 @@ impl AsHandle for RecvBuffer {
 impl RecvBuffer {
     #[inline]
     pub(crate) fn addr(&self) -> usize {
-        self.storage.as_ptr().expose_addr() + self.offset
+        self.storage.as_ptr().addr() + self.offset
     }
 
     #[inline]