liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

executor work

Browse Source
This commit is contained in:
John Smith
2022-06-27 23:46:29 -04:00
parent ebea72c9db
commit fdbb4c6397
59 changed files with 726 additions and 640 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
veilid-core/src/xx/async_peek_stream.rs
View File

@@ -5,17 +5,8 @@ use task::{Context, Poll};
////////
///
trait SendStream: AsyncRead + AsyncWrite + Send + Unpin {
fn clone_stream(&self) -> Box<dyn SendStream>;
}
impl<S> SendStream for S
where
S: AsyncRead + AsyncWrite + Send + Clone + Unpin + 'static,
{
fn clone_stream(&self) -> Box<dyn SendStream> {
Box::new(self.clone())
}
}
trait SendStream: AsyncRead + AsyncWrite + Send + Unpin {}
impl<S> SendStream for S where S: AsyncRead + AsyncWrite + Send + Unpin + 'static {}
////////
///
@@ -126,7 +117,7 @@ where
impl AsyncPeekStream {
pub fn new<S>(stream: S) -> Self
where
S: AsyncRead + AsyncWrite + Send + Clone + Unpin + 'static,
S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
Self {
inner: Arc::new(Mutex::new(AsyncPeekStreamInner {

18
veilid-core/src/xx/mod.rs
View File

@@ -11,7 +11,7 @@ mod log_thru;
mod must_join_handle;
mod must_join_single_future;
mod mutable_future;
mod single_future;
// mod single_future;
mod single_shot_eventual;
mod split_url;
mod tick_task;
@@ -67,6 +67,7 @@ cfg_if! {
pub use no_std_net::{ SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs, IpAddr, Ipv4Addr, Ipv6Addr };
pub type SystemPinBoxFuture<T> = PinBox<dyn Future<Output = T> + 'static>;
pub type SystemPinBoxFutureLifetime<'a, T> = PinBox<dyn Future<Output = T> + 'a>;
pub use async_executors::JoinHandle as LowLevelJoinHandle;
} else {
pub use std::string::String;
pub use std::vec::Vec;
@@ -91,8 +92,17 @@ cfg_if! {
pub use std::time::Duration;
pub use std::pin::Pin;
pub use std::ops::{FnOnce, FnMut, Fn};
pub use async_std::sync::Mutex as AsyncMutex;
pub use async_std::sync::MutexGuard as AsyncMutexGuard;
cfg_if! {
if #[cfg(feature="rt-async-std")] {
pub use async_std::sync::Mutex as AsyncMutex;
pub use async_std::sync::MutexGuard as AsyncMutexGuard;
pub use async_std::task::JoinHandle as LowLevelJoinHandle;
} else if #[cfg(feature="rt-tokio")] {
pub use tokio::sync::Mutex as AsyncMutex;
pub use tokio::sync::MutexGuard as AsyncMutexGuard;
pub use tokio::task::JoinHandle as LowLevelJoinHandle;
}
}
pub use std::net::{ SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs, IpAddr, Ipv4Addr, Ipv6Addr };
pub type SystemPinBoxFuture<T> = PinBox<dyn Future<Output = T> + Send + 'static>;
pub type SystemPinBoxFutureLifetime<'a, T> = PinBox<dyn Future<Output = T> + Send + 'a>;
@@ -111,7 +121,7 @@ pub use ip_extra::*;
pub use must_join_handle::*;
pub use must_join_single_future::*;
pub use mutable_future::*;
pub use single_future::*;
// pub use single_future::*;
pub use single_shot_eventual::*;
pub use tick_task::*;
pub use tools::*;

37
veilid-core/src/xx/must_join_handle.rs
View File

@@ -1,21 +1,40 @@
use async_executors::JoinHandle;
use super::*;
use core::future::Future;
use core::pin::Pin;
use core::task::{Context, Poll};
#[derive(Debug)]
pub struct MustJoinHandle<T> {
join_handle: JoinHandle<T>,
join_handle: Option<LowLevelJoinHandle<T>>,
completed: bool,
}
impl<T> MustJoinHandle<T> {
pub fn new(join_handle: JoinHandle<T>) -> Self {
pub fn new(join_handle: LowLevelJoinHandle<T>) -> Self {
Self {
join_handle,
join_handle: Some(join_handle),
completed: false,
}
}
pub async fn abort(mut self) {
if !self.completed {
cfg_if! {
if #[cfg(feature="rt-async-std")] {
if let Some(jh) = self.join_handle.take() {
jh.cancel().await;
self.completed = true;
}
} else if #[cfg(feature="rt-tokio")] {
if let Some(jh) = self.join_handle.take() {
jh.abort();
let _ = jh.await;
self.completed = true;
}
}
}
}
}
}
impl<T> Drop for MustJoinHandle<T> {
@@ -31,10 +50,16 @@ impl<T: 'static> Future for MustJoinHandle<T> {
type Output = T;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match Pin::new(&mut self.join_handle).poll(cx) {
match Pin::new(self.join_handle.as_mut().unwrap()).poll(cx) {
Poll::Ready(t) => {
self.completed = true;
Poll::Ready(t)
cfg_if! {
if #[cfg(feature="rt-async-std")] {
Poll::Ready(t)
} else if #[cfg(feature="rt-tokio")] {
Poll::Ready(t.unwrap())
}
}
}
Poll::Pending => Poll::Pending,
}

7
veilid-core/src/xx/must_join_single_future.rs
View File

@@ -1,6 +1,5 @@
use super::*;
use crate::intf::*;
use cfg_if::*;
use crate::*;
use core::task::Poll;
use futures_util::poll;
@@ -160,7 +159,7 @@ where
// Run if we should do that
if run {
self.unlock(Some(MustJoinHandle::new(spawn_local(future))));
self.unlock(Some(intf::spawn_local(future)));
}
// Return the prior result if we have one
@@ -203,7 +202,7 @@ cfg_if! {
}
// Run if we should do that
if run {
self.unlock(Some(MustJoinHandle::new(spawn(future))));
self.unlock(Some(intf::spawn_with_local_set(future)));
}
// Return the prior result if we have one
Ok((out, run))

242
veilid-core/src/xx/single_future.rs
View File

@@ -1,242 +0,0 @@
use super::*;
use crate::intf::*;
use cfg_if::*;
use core::task::Poll;
use futures_util::poll;
#[derive(Debug)]
struct SingleFutureInner<T>
where
T: 'static,
{
locked: bool,
join_handle: Option<JoinHandle<T>>,
}
/// Spawns a single background processing task idempotently, possibly returning the return value of the previously executed background task
/// This does not queue, just ensures that no more than a single copy of the task is running at a time, but allowing tasks to be retriggered
#[derive(Debug, Clone)]
pub struct SingleFuture<T>
where
T: 'static,
{
inner: Arc<Mutex<SingleFutureInner<T>>>,
}
impl<T> Default for SingleFuture<T>
where
T: 'static,
{
fn default() -> Self {
Self::new()
}
}
impl<T> SingleFuture<T>
where
T: 'static,
{
pub fn new() -> Self {
Self {
inner: Arc::new(Mutex::new(SingleFutureInner {
locked: false,
join_handle: None,
})),
}
}
fn try_lock(&self) -> Result<Option<JoinHandle<T>>, ()> {
let mut inner = self.inner.lock();
if inner.locked {
// If already locked error out
return Err(());
}
inner.locked = true;
// If we got the lock, return what we have for a join handle if anything
Ok(inner.join_handle.take())
}
fn unlock(&self, jh: Option<JoinHandle<T>>) {
let mut inner = self.inner.lock();
assert!(inner.locked);
assert!(inner.join_handle.is_none());
inner.locked = false;
inner.join_handle = jh;
}
// Check the result
pub async fn check(&self) -> Result<Option<T>, ()> {
let mut out: Option<T> = None;
// See if we have a result we can return
let maybe_jh = match self.try_lock() {
Ok(v) => v,
Err(_) => {
// If we are already polling somewhere else, don't hand back a result
return Err(());
}
};
if maybe_jh.is_some() {
let mut jh = maybe_jh.unwrap();
// See if we finished, if so, return the value of the last execution
if let Poll::Ready(r) = poll!(&mut jh) {
out = Some(r);
// Task finished, unlock with nothing
self.unlock(None);
} else {
// Still running put the join handle back so we can check on it later
self.unlock(Some(jh));
}
} else {
// No task, unlock with nothing
self.unlock(None);
}
// Return the prior result if we have one
Ok(out)
}
// Wait for the result
pub async fn join(&self) -> Result<Option<T>, ()> {
let mut out: Option<T> = None;
// See if we have a result we can return
let maybe_jh = match self.try_lock() {
Ok(v) => v,
Err(_) => {
// If we are already polling somewhere else,
// that's an error because you can only join
// these things once
return Err(());
}
};
if maybe_jh.is_some() {
let jh = maybe_jh.unwrap();
// Wait for return value of the last execution
out = Some(jh.await);
// Task finished, unlock with nothing
} else {
// No task, unlock with nothing
}
self.unlock(None);
// Return the prior result if we have one
Ok(out)
}
// Cancel
pub async fn cancel(&self) -> Result<Option<T>, ()> {
let mut out: Option<T> = None;
// See if we have a result we can return
let maybe_jh = match self.try_lock() {
Ok(v) => v,
Err(_) => {
// If we are already polling somewhere else, don't hand back a result
return Err(());
}
};
if maybe_jh.is_some() {
let mut jh = maybe_jh.unwrap();
// See if we finished, if so, return the value of the last execution
if let Poll::Ready(r) = poll!(&mut jh) {
out = Some(r);
// Task finished, unlock with nothing
} else {
// Still running but drop the join handle anyway to cancel the task, unlock with nothing
}
}
self.unlock(None);
// Return the prior result if we have one
Ok(out)
}
// Possibly spawn the future possibly returning the value of the last execution
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
pub async fn single_spawn(
&self,
future: impl Future<Output = T> + 'static,
) -> Result<(Option<T>, bool), ()> {
let mut out: Option<T> = None;
// See if we have a result we can return
let maybe_jh = match self.try_lock() {
Ok(v) => v,
Err(_) => {
// If we are already polling somewhere else, don't hand back a result
return Err(());
}
};
let mut run = true;
if maybe_jh.is_some() {
let mut jh = maybe_jh.unwrap();
// See if we finished, if so, return the value of the last execution
if let Poll::Ready(r) = poll!(&mut jh) {
out = Some(r);
// Task finished, unlock with a new task
} else {
// Still running, don't run again, unlock with the current join handle
run = false;
self.unlock(Some(jh));
}
}
// Run if we should do that
if run {
self.unlock(Some(spawn_local(future)));
}
// Return the prior result if we have one
Ok((out, run))
}
}
}
}
cfg_if! {
if #[cfg(not(target_arch = "wasm32"))] {
impl<T> SingleFuture<T>
where
T: 'static + Send,
{
pub async fn single_spawn(
&self,
future: impl Future<Output = T> + Send + 'static,
) -> Result<(Option<T>, bool), ()> {
let mut out: Option<T> = None;
// See if we have a result we can return
let maybe_jh = match self.try_lock() {
Ok(v) => v,
Err(_) => {
// If we are already polling somewhere else, don't hand back a result
return Err(());
}
};
let mut run = true;
if maybe_jh.is_some() {
let mut jh = maybe_jh.unwrap();
// See if we finished, if so, return the value of the last execution
if let Poll::Ready(r) = poll!(&mut jh) {
out = Some(r);
// Task finished, unlock with a new task
} else {
// Still running, don't run again, unlock with the current join handle
run = false;
self.unlock(Some(jh));
}
}
// Run if we should do that
if run {
self.unlock(Some(spawn(future)));
}
// Return the prior result if we have one
Ok((out, run))
}
}
}
}

4
veilid-core/src/xx/tick_task.rs
View File

@@ -1,5 +1,5 @@
use super::*;
use crate::intf::*;
use crate::*;
use core::sync::atomic::{AtomicU64, Ordering};
use once_cell::sync::OnceCell;
@@ -90,7 +90,7 @@ impl TickTask {
}
pub async fn tick(&self) -> Result<(), String> {
let now = get_timestamp();
let now = intf::get_timestamp();
let last_timestamp_us = self.last_timestamp_us.load(Ordering::Acquire);
if last_timestamp_us != 0u64 && (now - last_timestamp_us) < self.tick_period_us {