From 06915ec25dbda29220ca9dcdb83f7b54db8937c7 Mon Sep 17 00:00:00 2001 From: Sebastien Bourdeauducq Date: Sun, 7 Jun 2020 21:47:36 +0800 Subject: [PATCH] rpc: make alloc async --- src/runtime/src/rpc.rs | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/src/runtime/src/rpc.rs b/src/runtime/src/rpc.rs index ff7aa86b..6440cee1 100644 --- a/src/runtime/src/rpc.rs +++ b/src/runtime/src/rpc.rs @@ -1,4 +1,5 @@ use core::str; +use core::future::Future; use cslice::{CSlice, CMutSlice}; use log::debug; @@ -13,9 +14,10 @@ use crate::proto_async; use self::tag::{Tag, TagIterator, split_tag}; #[async_recursion(?Send)] -async unsafe fn recv_value(stream: &TcpStream, tag: Tag<'async_recursion>, data: &mut *mut (), - alloc: &dyn Fn(usize) -> *mut ()) - -> Result<(), smoltcp::Error> +async unsafe fn recv_value(stream: &TcpStream, tag: Tag<'async_recursion>, data: &mut *mut (), + alloc: &(impl Fn(usize) -> F + 'async_recursion)) + -> Result<(), smoltcp::Error> + where F: Future { macro_rules! consume_value { ($ty:ty, |$ptr:ident| $map:expr) => ({ @@ -45,7 +47,7 @@ async unsafe fn recv_value(stream: &TcpStream, tag: Tag<'async_recursion>, data: Tag::String | Tag::Bytes | Tag::ByteArray => { consume_value!(CMutSlice, |ptr| { let length = proto_async::read_i32(stream).await? as usize; - *ptr = CMutSlice::new(alloc(length) as *mut u8, length); + *ptr = CMutSlice::new(alloc(length).await as *mut u8, length); proto_async::read_chunk(stream, (*ptr).as_mut()).await?; Ok(()) }) @@ -64,7 +66,7 @@ async unsafe fn recv_value(stream: &TcpStream, tag: Tag<'async_recursion>, data: (*ptr).length = proto_async::read_i32(stream).await? as u32; let tag = it.clone().next().expect("truncated tag"); - (*ptr).elements = alloc(tag.size() * (*ptr).length as usize); + (*ptr).elements = alloc(tag.size() * (*ptr).length as usize).await; let mut data = (*ptr).elements; for _ in 0..(*ptr).length as usize { @@ -85,9 +87,10 @@ async unsafe fn recv_value(stream: &TcpStream, tag: Tag<'async_recursion>, data: } } -pub async fn recv_return(stream: &TcpStream, tag_bytes: &[u8], data: *mut (), - alloc: &dyn Fn(usize) -> *mut ()) - -> Result<(), smoltcp::Error> +pub async fn recv_return(stream: &TcpStream, tag_bytes: &[u8], data: *mut (), + alloc: &impl Fn(usize) -> F) + -> Result<(), smoltcp::Error> + where F: Future { let mut it = TagIterator::new(tag_bytes); debug!("recv ...->{}", it);