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

refactor

Browse Source
This commit is contained in:
John Smith
2022-11-26 21:37:23 -05:00
parent 5df46aecae
commit b1bdf76ae8
80 changed files with 865 additions and 700 deletions
Download Patch File Download Diff File Expand all files Collapse all files

222
veilid-tools/src/async_peek_stream.rs Normal file
View File

@@ -0,0 +1,222 @@
use super::*;
use std::io;
use task::{Context, Poll};
////////
///
trait SendStream: AsyncRead + AsyncWrite + Send + Unpin {}
impl<S> SendStream for S where S: AsyncRead + AsyncWrite + Send + Unpin + 'static {}
////////
///
pub struct Peek<'a> {
aps: AsyncPeekStream,
buf: &'a mut [u8],
}
impl Unpin for Peek<'_> {}
impl Future for Peek<'_> {
type Output = std::io::Result<usize>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = &mut *self;
let mut inner = this.aps.inner.lock();
let inner = &mut *inner;
//
let buf_len = this.buf.len();
let mut copy_len = buf_len;
if buf_len > inner.peekbuf_len {
//
inner.peekbuf.resize(buf_len, 0u8);
let read_len = match Pin::new(&mut inner.stream).poll_read(
cx,
&mut inner.peekbuf.as_mut_slice()[inner.peekbuf_len..buf_len],
) {
Poll::Pending => {
inner.peekbuf.resize(inner.peekbuf_len, 0u8);
return Poll::Pending;
}
Poll::Ready(Err(e)) => {
return Poll::Ready(Err(e));
}
Poll::Ready(Ok(v)) => v,
};
inner.peekbuf_len += read_len;
inner.peekbuf.resize(inner.peekbuf_len, 0u8);
copy_len = inner.peekbuf_len;
}
this.buf[..copy_len].copy_from_slice(&inner.peekbuf[..copy_len]);
Poll::Ready(Ok(copy_len))
}
}
////////
///
pub struct PeekExact<'a> {
aps: AsyncPeekStream,
buf: &'a mut [u8],
cur_read: usize,
}
impl Unpin for PeekExact<'_> {}
impl Future for PeekExact<'_> {
type Output = std::io::Result<usize>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = &mut *self;
let mut inner = this.aps.inner.lock();
let inner = &mut *inner;
//
let buf_len = this.buf.len();
let mut copy_len = buf_len;
if buf_len > inner.peekbuf_len {
//
inner.peekbuf.resize(buf_len, 0u8);
let read_len = match Pin::new(&mut inner.stream).poll_read(
cx,
&mut inner.peekbuf.as_mut_slice()[inner.peekbuf_len..buf_len],
) {
Poll::Pending => {
inner.peekbuf.resize(inner.peekbuf_len, 0u8);
return Poll::Pending;
}
Poll::Ready(Err(e)) => {
return Poll::Ready(Err(e));
}
Poll::Ready(Ok(v)) => v,
};
inner.peekbuf_len += read_len;
inner.peekbuf.resize(inner.peekbuf_len, 0u8);
copy_len = inner.peekbuf_len;
}
this.buf[this.cur_read..copy_len].copy_from_slice(&inner.peekbuf[this.cur_read..copy_len]);
this.cur_read = copy_len;
if this.cur_read == buf_len {
Poll::Ready(Ok(buf_len))
} else {
Poll::Pending
}
}
}
/////////
///
struct AsyncPeekStreamInner {
stream: Box<dyn SendStream>,
peekbuf: Vec<u8>,
peekbuf_len: usize,
}
#[derive(Clone)]
pub struct AsyncPeekStream
where
Self: AsyncRead + AsyncWrite + Send + Unpin,
{
inner: Arc<Mutex<AsyncPeekStreamInner>>,
}
impl AsyncPeekStream {
pub fn new<S>(stream: S) -> Self
where
S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
Self {
inner: Arc::new(Mutex::new(AsyncPeekStreamInner {
stream: Box::new(stream),
peekbuf: Vec::new(),
peekbuf_len: 0,
})),
}
}
pub fn peek<'a>(&'a self, buf: &'a mut [u8]) -> Peek<'a> {
Peek::<'a> {
aps: self.clone(),
buf,
}
}
pub fn peek_exact<'a>(&'a self, buf: &'a mut [u8]) -> PeekExact<'a> {
PeekExact::<'a> {
aps: self.clone(),
buf,
cur_read: 0,
}
}
}
impl AsyncRead for AsyncPeekStream {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
let mut inner = self.inner.lock();
//
let buflen = buf.len();
let bufcopylen = core::cmp::min(buflen, inner.peekbuf_len);
let bufreadlen = if buflen > inner.peekbuf_len {
buflen - inner.peekbuf_len
} else {
0
};
if bufreadlen > 0 {
match Pin::new(&mut inner.stream).poll_read(cx, &mut buf[bufcopylen..buflen]) {
Poll::Ready(res) => {
let readlen = res?;
buf[0..bufcopylen].copy_from_slice(&inner.peekbuf[0..bufcopylen]);
inner.peekbuf_len = 0;
Poll::Ready(Ok(bufcopylen + readlen))
}
Poll::Pending => {
if bufcopylen > 0 {
buf[0..bufcopylen].copy_from_slice(&inner.peekbuf[0..bufcopylen]);
inner.peekbuf_len = 0;
Poll::Ready(Ok(bufcopylen))
} else {
Poll::Pending
}
}
}
} else {
buf[0..bufcopylen].copy_from_slice(&inner.peekbuf[0..bufcopylen]);
if bufcopylen == inner.peekbuf_len {
inner.peekbuf_len = 0;
} else if bufcopylen != 0 {
// slide buffer over by bufcopylen
let tail = inner.peekbuf.split_off(bufcopylen);
inner.peekbuf = tail;
inner.peekbuf_len -= bufcopylen;
}
Poll::Ready(Ok(bufcopylen))
}
}
}
impl AsyncWrite for AsyncPeekStream {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let mut inner = self.inner.lock();
Pin::new(&mut inner.stream).poll_write(cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let mut inner = self.inner.lock();
Pin::new(&mut inner.stream).poll_flush(cx)
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let mut inner = self.inner.lock();
Pin::new(&mut inner.stream).poll_close(cx)
}
}
impl core::marker::Unpin for AsyncPeekStream {}

138
veilid-tools/src/async_tag_lock.rs Normal file
View File

@@ -0,0 +1,138 @@
use super::*;
use core::fmt::Debug;
use core::hash::Hash;
#[derive(Debug)]
pub struct AsyncTagLockGuard<T>
where
T: Hash + Eq + Clone + Debug,
{
table: AsyncTagLockTable<T>,
tag: T,
_guard: AsyncMutexGuardArc<()>,
}
impl<T> AsyncTagLockGuard<T>
where
T: Hash + Eq + Clone + Debug,
{
fn new(table: AsyncTagLockTable<T>, tag: T, guard: AsyncMutexGuardArc<()>) -> Self {
Self {
table,
tag,
_guard: guard,
}
}
}
impl<T> Drop for AsyncTagLockGuard<T>
where
T: Hash + Eq + Clone + Debug,
{
fn drop(&mut self) {
let mut inner = self.table.inner.lock();
// Inform the table we're dropping this guard
let waiters = {
// Get the table entry, it must exist since we have a guard locked
let entry = inner.table.get_mut(&self.tag).unwrap();
// Decrement the number of waiters
entry.waiters -= 1;
// Return the number of waiters left
entry.waiters
};
// If there are no waiters left, we remove the tag from the table
if waiters == 0 {
inner.table.remove(&self.tag).unwrap();
}
// Proceed with releasing _guard, which may cause some concurrent tag lock to acquire
}
}
#[derive(Clone, Debug)]
struct AsyncTagLockTableEntry {
mutex: Arc<AsyncMutex<()>>,
waiters: usize,
}
struct AsyncTagLockTableInner<T>
where
T: Hash + Eq + Clone + Debug,
{
table: HashMap<T, AsyncTagLockTableEntry>,
}
#[derive(Clone)]
pub struct AsyncTagLockTable<T>
where
T: Hash + Eq + Clone + Debug,
{
inner: Arc<Mutex<AsyncTagLockTableInner<T>>>,
}
impl<T> fmt::Debug for AsyncTagLockTable<T>
where
T: Hash + Eq + Clone + Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("AsyncTagLockTable").finish()
}
}
impl<T> AsyncTagLockTable<T>
where
T: Hash + Eq + Clone + Debug,
{
pub fn new() -> Self {
Self {
inner: Arc::new(Mutex::new(AsyncTagLockTableInner {
table: HashMap::new(),
})),
}
}
pub fn len(&self) -> usize {
let inner = self.inner.lock();
inner.table.len()
}
pub async fn lock_tag(&self, tag: T) -> AsyncTagLockGuard<T> {
// Get or create a tag lock entry
let mutex = {
let mut inner = self.inner.lock();
// See if this tag is in the table
// and if not, add a new mutex for this tag
let entry = inner
.table
.entry(tag.clone())
.or_insert_with(|| AsyncTagLockTableEntry {
mutex: Arc::new(AsyncMutex::new(())),
waiters: 0,
});
// Increment the number of waiters
entry.waiters += 1;
// Return the mutex associated with the tag
entry.mutex.clone()
// Drop the table guard
};
// Lock the tag lock
let guard;
cfg_if! {
if #[cfg(feature="rt-tokio")] {
// tokio version
guard = mutex.lock_owned().await;
} else {
// async-std and wasm async-lock version
guard = mutex.lock_arc().await;
}
}
// Return the locked guard
AsyncTagLockGuard::new(self.clone(), tag, guard)
}
}

108
veilid-tools/src/bump_port.rs Normal file
View File

@@ -0,0 +1,108 @@
use super::*;
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
} else {
use std::net::{TcpListener, UdpSocket};
}
}
#[derive(ThisError, Debug, Clone, PartialEq, Eq)]
pub enum BumpPortError {
#[error("Unsupported architecture")]
Unsupported,
#[error("Failure: {0}")]
Failed(String),
}
pub enum BumpPortType {
UDP,
TCP,
}
pub fn tcp_port_available(addr: &SocketAddr) -> bool {
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
true
} else {
match TcpListener::bind(addr) {
Ok(_) => true,
Err(_) => false,
}
}
}
}
pub fn udp_port_available(addr: &SocketAddr) -> bool {
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
true
} else {
match UdpSocket::bind(addr) {
Ok(_) => true,
Err(_) => false,
}
}
}
}
pub fn bump_port(addr: &mut SocketAddr, bpt: BumpPortType) -> Result<bool, BumpPortError> {
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
Err(BumpPortError::Unsupported)
}
else
{
let mut bumped = false;
let mut port = addr.port();
let mut addr_bump = addr.clone();
loop {
if match bpt {
BumpPortType::TCP => tcp_port_available(&addr_bump),
BumpPortType::UDP => udp_port_available(&addr_bump),
} {
*addr = addr_bump;
return Ok(bumped);
}
if port == u16::MAX {
break;
}
port += 1;
addr_bump.set_port(port);
bumped = true;
}
Err(BumpPortError::Failure("no ports remaining".to_owned()))
}
}
}
pub fn bump_port_string(addr: &mut String, bpt: BumpPortType) -> Result<bool, BumpPortError> {
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
return Err(BumpPortError::Unsupported);
}
else
{
let savec: Vec<SocketAddr> = addr
.to_socket_addrs()
.map_err(|x| BumpPortError::Failure(format!("failed to resolve socket address: {}", x)))?
.collect();
if savec.len() == 0 {
return Err(BumpPortError::Failure("No socket addresses resolved".to_owned()));
}
let mut sa = savec.first().unwrap().clone();
if !bump_port(&mut sa, bpt)? {
return Ok(false);
}
*addr = sa.to_string();
Ok(true)
}
}
}

112
veilid-tools/src/clone_stream.rs Normal file
View File

@@ -0,0 +1,112 @@
use super::*;
use core::pin::Pin;
use core::task::{Context, Poll};
use futures_util::AsyncRead as Read;
use futures_util::AsyncWrite as Write;
use futures_util::Sink;
use futures_util::Stream;
use std::io;
pub struct CloneStream<T>
where
T: Unpin,
{
inner: Arc<Mutex<T>>,
}
impl<T> Clone for CloneStream<T>
where
T: Unpin,
{
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T> CloneStream<T>
where
T: Unpin,
{
pub fn new(t: T) -> Self {
Self {
inner: Arc::new(Mutex::new(t)),
}
}
}
impl<T> Read for CloneStream<T>
where
T: Read + Unpin,
{
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_read(cx, buf)
}
}
impl<T> Write for CloneStream<T>
where
T: Write + Unpin,
{
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_write(cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_flush(cx)
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_close(cx)
}
}
impl<T> Stream for CloneStream<T>
where
T: Stream + Unpin,
{
type Item = T::Item;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_next(cx)
}
}
impl<T, Item> Sink<Item> for CloneStream<T>
where
T: Sink<Item> + Unpin,
{
type Error = T::Error;
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_ready(cx)
}
fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), Self::Error> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).start_send(item)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_flush(cx)
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut inner = self.inner.lock();
Pin::new(&mut *inner).poll_close(cx)
}
}

219
veilid-tools/src/eventual.rs Normal file
View File

@@ -0,0 +1,219 @@
use super::*;
use eventual_base::*;
pub struct Eventual {
inner: Arc<Mutex<EventualBaseInner<()>>>,
}
impl core::fmt::Debug for Eventual {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Eventual").finish()
}
}
impl Clone for Eventual {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl EventualBase for Eventual {
type ResolvedType = ();
fn base_inner(&self) -> MutexGuard<EventualBaseInner<Self::ResolvedType>> {
self.inner.lock()
}
}
impl Default for Eventual {
fn default() -> Self {
Self::new()
}
}
impl Eventual {
pub fn new() -> Self {
Self {
inner: Arc::new(Mutex::new(EventualBaseInner::new())),
}
}
pub fn instance_clone<T>(&self, value: T) -> EventualFutureClone<T>
where
T: Clone + Unpin,
{
EventualFutureClone {
id: None,
value,
eventual: self.clone(),
}
}
pub fn instance_none<T>(&self) -> EventualFutureNone<T>
where
T: Unpin,
{
EventualFutureNone {
id: None,
eventual: self.clone(),
_marker: core::marker::PhantomData {},
}
}
pub fn instance_empty(&self) -> EventualFutureEmpty {
EventualFutureEmpty {
id: None,
eventual: self.clone(),
}
}
pub fn resolve(&self) -> EventualResolvedFuture<Self> {
self.resolve_to_value(())
}
}
///////
pub struct EventualFutureClone<T>
where
T: Clone + Unpin,
{
id: Option<usize>,
value: T,
eventual: Eventual,
}
impl<T> Future for EventualFutureClone<T>
where
T: Clone + Unpin,
{
type Output = T;
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let this = &mut *self;
let out = {
let mut inner = this.eventual.base_inner();
inner.instance_poll(&mut this.id, cx)
};
match out {
None => task::Poll::<Self::Output>::Pending,
Some(wakers) => {
// Wake all EventualResolvedFutures
for w in wakers {
w.wake();
}
task::Poll::<Self::Output>::Ready(this.value.clone())
}
}
}
}
impl<T> Drop for EventualFutureClone<T>
where
T: Clone + Unpin,
{
fn drop(&mut self) {
if let Some(id) = self.id.take() {
let wakers = {
let mut inner = self.eventual.base_inner();
inner.remove_waker(id)
};
for w in wakers {
w.wake();
}
}
}
}
///////
pub struct EventualFutureNone<T>
where
T: Unpin,
{
id: Option<usize>,
eventual: Eventual,
_marker: core::marker::PhantomData<T>,
}
impl<T> Future for EventualFutureNone<T>
where
T: Unpin,
{
type Output = Option<T>;
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let this = &mut *self;
let out = {
let mut inner = this.eventual.base_inner();
inner.instance_poll(&mut this.id, cx)
};
match out {
None => task::Poll::<Self::Output>::Pending,
Some(wakers) => {
// Wake all EventualResolvedFutures
for w in wakers {
w.wake();
}
task::Poll::<Self::Output>::Ready(None)
}
}
}
}
impl<T> Drop for EventualFutureNone<T>
where
T: Unpin,
{
fn drop(&mut self) {
if let Some(id) = self.id.take() {
let wakers = {
let mut inner = self.eventual.base_inner();
inner.remove_waker(id)
};
for w in wakers {
w.wake();
}
}
}
}
///////
pub struct EventualFutureEmpty {
id: Option<usize>,
eventual: Eventual,
}
impl Future for EventualFutureEmpty {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let this = &mut *self;
let out = {
let mut inner = this.eventual.base_inner();
inner.instance_poll(&mut this.id, cx)
};
match out {
None => task::Poll::<Self::Output>::Pending,
Some(wakers) => {
// Wake all EventualResolvedFutures
for w in wakers {
w.wake();
}
task::Poll::<Self::Output>::Ready(())
}
}
}
}
impl Drop for EventualFutureEmpty {
fn drop(&mut self) {
if let Some(id) = self.id.take() {
let wakers = {
let mut inner = self.eventual.base_inner();
inner.remove_waker(id)
};
for w in wakers {
w.wake();
}
}
}
}

217
veilid-tools/src/eventual_base.rs Normal file
View File

@@ -0,0 +1,217 @@
use super::*;
#[derive(ThisError, Debug, Clone, PartialEq, Eq)]
pub enum EventualError {
#[error("Try failed: {0}")]
TryFailed(String),
}
pub struct EventualBaseInner<T> {
resolved: Option<T>,
wakers: BTreeMap<usize, task::Waker>,
resolved_wakers: BTreeMap<usize, task::Waker>,
freelist: Vec<usize>,
resolved_freelist: Vec<usize>,
}
impl<T> EventualBaseInner<T> {
pub(super) fn new() -> Self {
EventualBaseInner {
resolved: None,
wakers: BTreeMap::new(),
resolved_wakers: BTreeMap::new(),
freelist: Vec::new(),
resolved_freelist: Vec::new(),
}
}
pub(super) fn insert_waker(&mut self, waker: task::Waker) -> usize {
let id = match self.freelist.pop() {
Some(id) => id,
None => self.wakers.len(),
};
self.wakers.insert(id, waker);
id
}
#[must_use]
pub(super) fn remove_waker(&mut self, id: usize) -> Vec<task::Waker> {
self.freelist.push(id);
self.wakers.remove(&id);
// See if we should complete the EventualResolvedFutures
let mut resolved_waker_list = Vec::new();
if self.wakers.is_empty() && self.resolved.is_some() {
for w in &self.resolved_wakers {
resolved_waker_list.push(w.1.clone());
}
}
resolved_waker_list
}
pub(super) fn insert_resolved_waker(&mut self, waker: task::Waker) -> usize {
let id = match self.resolved_freelist.pop() {
Some(id) => id,
None => self.resolved_wakers.len(),
};
self.resolved_wakers.insert(id, waker);
id
}
pub(super) fn remove_resolved_waker(&mut self, id: usize) {
self.resolved_freelist.push(id);
self.resolved_wakers.remove(&id);
}
#[must_use]
pub(super) fn resolve_and_get_wakers(&mut self, value: T) -> Option<Vec<task::Waker>> {
if self.resolved.is_some() {
// Already resolved
return None;
}
// Store resolved value
self.resolved = Some(value);
// Return a copy of the waker list so the caller can wake all the EventualFutures
let mut waker_list = Vec::new();
for w in &self.wakers {
waker_list.push(w.1.clone());
}
Some(waker_list)
}
pub(super) fn is_resolved(&self) -> bool {
self.resolved.is_some()
}
pub(super) fn resolved_value_ref(&self) -> &Option<T> {
&self.resolved
}
pub(super) fn resolved_value_mut(&mut self) -> &mut Option<T> {
&mut self.resolved
}
pub(super) fn reset(&mut self) {
assert_eq!(self.wakers.len(), 0);
assert_eq!(self.resolved_wakers.len(), 0);
self.resolved = None;
self.freelist.clear();
self.resolved_freelist.clear();
}
pub(super) fn try_reset(&mut self) -> Result<(), EventualError> {
if !self.wakers.is_empty() {
return Err(EventualError::TryFailed(
"wakers not empty during reset".to_owned(),
));
}
if !self.resolved_wakers.is_empty() {
return Err(EventualError::TryFailed(
"Resolved wakers not empty during reset".to_owned(),
));
}
self.reset();
Ok(())
}
// Resolved future helpers
pub(super) fn resolved_poll(
&mut self,
id: &mut Option<usize>,
cx: &mut task::Context<'_>,
) -> task::Poll<()> {
// If there are any instance futures still waiting, we resolution isn't finished
if !self.wakers.is_empty() {
if id.is_none() {
*id = Some(self.insert_resolved_waker(cx.waker().clone()));
}
task::Poll::<()>::Pending
} else {
if let Some(id) = id.take() {
self.remove_resolved_waker(id);
}
task::Poll::<()>::Ready(())
}
}
// Instance future helpers
#[must_use]
pub(super) fn instance_poll(
&mut self,
id: &mut Option<usize>,
cx: &mut task::Context<'_>,
) -> Option<Vec<task::Waker>> {
// If the resolved value hasn't showed up then we can't wake the instance futures
if self.resolved.is_none() {
if id.is_none() {
*id = Some(self.insert_waker(cx.waker().clone()));
}
None
} else if let Some(id) = id.take() {
Some(self.remove_waker(id))
} else {
Some(Vec::new())
}
}
}
// xxx: this would love to be 'pub(super)' instead of pub, to ensure nobody else touches resolve_to_value directly
pub trait EventualBase: Clone + Unpin {
type ResolvedType;
fn base_inner(&self) -> MutexGuard<EventualBaseInner<Self::ResolvedType>>;
fn resolve_to_value(&self, value: Self::ResolvedType) -> EventualResolvedFuture<Self> {
let wakers = {
let mut inner = self.base_inner();
inner.resolve_and_get_wakers(value)
};
if let Some(wakers) = wakers {
for w in wakers {
w.wake();
}
}
EventualResolvedFuture {
id: None,
eventual: self.clone(),
}
}
}
pub struct EventualResolvedFuture<B: EventualBase> {
id: Option<usize>,
eventual: B,
}
impl<B: EventualBase> Future for EventualResolvedFuture<B> {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let this = &mut *self;
let mut inner = this.eventual.base_inner();
inner.resolved_poll(&mut this.id, cx)
}
}
impl<B: EventualBase> Drop for EventualResolvedFuture<B> {
fn drop(&mut self) {
if let Some(id) = self.id.take() {
let mut inner = self.eventual.base_inner();
inner.remove_resolved_waker(id);
}
}
}
pub trait EventualCommon: EventualBase {
fn is_resolved(&self) -> bool {
self.base_inner().is_resolved()
}
fn reset(&self) {
self.base_inner().reset()
}
fn try_reset(&self) -> Result<(), EventualError> {
self.base_inner().try_reset()
}
}
impl<T> EventualCommon for T where T: EventualBase {}

109
veilid-tools/src/eventual_value.rs Normal file
View File

@@ -0,0 +1,109 @@
use super::*;
use eventual_base::*;
pub struct EventualValue<T: Unpin> {
inner: Arc<Mutex<EventualBaseInner<T>>>,
}
impl<T: Unpin> core::fmt::Debug for EventualValue<T> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("EventualValue").finish()
}
}
impl<T: Unpin> Clone for EventualValue<T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T: Unpin> EventualBase for EventualValue<T> {
type ResolvedType = T;
fn base_inner(&self) -> MutexGuard<EventualBaseInner<Self::ResolvedType>> {
self.inner.lock()
}
}
impl<T: Unpin> Default for EventualValue<T> {
fn default() -> Self {
Self::new()
}
}
impl<T: Unpin> EventualValue<T> {
pub fn new() -> Self {
Self {
inner: Arc::new(Mutex::new(EventualBaseInner::new())),
}
}
pub fn instance(&self) -> EventualValueFuture<T> {
EventualValueFuture {
id: None,
eventual: self.clone(),
}
}
pub fn resolve(&self, value: T) -> EventualResolvedFuture<Self> {
self.resolve_to_value(value)
}
pub fn take_value(&self) -> Option<T> {
let mut inner = self.inner.lock();
inner.resolved_value_mut().take()
}
}
pub struct EventualValueFuture<T: Unpin> {
id: Option<usize>,
eventual: EventualValue<T>,
}
impl<T: Unpin> core::fmt::Debug for EventualValueFuture<T> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("EventualValueFuture")
.field("id", &self.id)
.finish()
}
}
impl<T: Unpin> Future for EventualValueFuture<T> {
type Output = EventualValue<T>;
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let this = &mut *self;
let out = {
let mut inner = this.eventual.base_inner();
inner.instance_poll(&mut this.id, cx)
};
match out {
None => task::Poll::<Self::Output>::Pending,
Some(wakers) => {
// Wake all EventualResolvedFutures
for w in wakers {
w.wake();
}
task::Poll::<Self::Output>::Ready(this.eventual.clone())
}
}
}
}
impl<T> Drop for EventualValueFuture<T>
where
T: Unpin,
{
fn drop(&mut self) {
if let Some(id) = self.id.take() {
let wakers = {
let mut inner = self.eventual.base_inner();
inner.remove_waker(id)
};
for w in wakers {
w.wake();
}
}
}
}

105
veilid-tools/src/eventual_value_clone.rs Normal file
View File

@@ -0,0 +1,105 @@
use super::*;
use eventual_base::*;
pub struct EventualValueClone<T: Unpin + Clone> {
inner: Arc<Mutex<EventualBaseInner<T>>>,
}
impl<T: Unpin + Clone> core::fmt::Debug for EventualValueClone<T> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("EventualValueClone").finish()
}
}
impl<T: Unpin + Clone> Clone for EventualValueClone<T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T: Unpin + Clone> EventualBase for EventualValueClone<T> {
type ResolvedType = T;
fn base_inner(&self) -> MutexGuard<EventualBaseInner<Self::ResolvedType>> {
self.inner.lock()
}
}
impl<T: Unpin + Clone> Default for EventualValueClone<T> {
fn default() -> Self {
Self::new()
}
}
impl<T: Unpin + Clone> EventualValueClone<T> {
pub fn new() -> Self {
Self {
inner: Arc::new(Mutex::new(EventualBaseInner::new())),
}
}
pub fn instance(&self) -> EventualValueCloneFuture<T>
where
T: Clone + Unpin,
{
EventualValueCloneFuture {
id: None,
eventual: self.clone(),
}
}
pub fn resolve(&self, value: T) -> EventualResolvedFuture<Self> {
self.resolve_to_value(value)
}
pub fn value(&self) -> Option<T> {
let inner = self.inner.lock();
inner.resolved_value_ref().clone()
}
}
pub struct EventualValueCloneFuture<T: Unpin + Clone> {
id: Option<usize>,
eventual: EventualValueClone<T>,
}
impl<T: Unpin + Clone> Future for EventualValueCloneFuture<T> {
type Output = T;
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let this = &mut *self;
let (out, some_value) = {
let mut inner = this.eventual.base_inner();
let out = inner.instance_poll(&mut this.id, cx);
(out, inner.resolved_value_ref().clone())
};
match out {
None => task::Poll::<Self::Output>::Pending,
Some(wakers) => {
// Wake all EventualResolvedFutures
for w in wakers {
w.wake();
}
task::Poll::<Self::Output>::Ready(some_value.unwrap())
}
}
}
}
impl<T> Drop for EventualValueCloneFuture<T>
where
T: Clone + Unpin,
{
fn drop(&mut self) {
if let Some(id) = self.id.take() {
let wakers = {
let mut inner = self.eventual.base_inner();
inner.remove_waker(id)
};
for w in wakers {
w.wake();
}
}
}
}

49
veilid-tools/src/interval.rs Normal file
View File

@@ -0,0 +1,49 @@
use super::*;
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
pub fn interval<F, FUT>(freq_ms: u32, callback: F) -> SendPinBoxFuture<()>
where
F: Fn() -> FUT + Send + Sync + 'static,
FUT: Future<Output = ()> + Send,
{
let e = Eventual::new();
let ie = e.clone();
let jh = spawn(Box::pin(async move {
while timeout(freq_ms, ie.instance_clone(())).await.is_err() {
callback().await;
}
}));
Box::pin(async move {
e.resolve().await;
jh.await;
})
}
} else {
pub fn interval<F, FUT>(freq_ms: u32, callback: F) -> SendPinBoxFuture<()>
where
F: Fn() -> FUT + Send + Sync + 'static,
FUT: Future<Output = ()> + Send,
{
let e = Eventual::new();
let ie = e.clone();
let jh = spawn(async move {
while timeout(freq_ms, ie.instance_clone(())).await.is_err() {
callback().await;
}
});
Box::pin(async move {
e.resolve().await;
jh.await;
})
}
}
}

66
veilid-tools/src/ip_addr_port.rs Normal file
View File

@@ -0,0 +1,66 @@
use super::*;
use core::fmt;
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug, PartialOrd, Ord)]
pub struct IpAddrPort {
addr: IpAddr,
port: u16,
}
impl IpAddrPort {
pub fn new(addr: IpAddr, port: u16) -> Self {
Self { addr, port }
}
pub fn addr(&self) -> &IpAddr {
&self.addr
}
pub fn port(&self) -> u16 {
self.port
}
pub fn set_addr(&mut self, new_addr: IpAddr) {
self.addr = new_addr;
}
pub fn set_port(&mut self, new_port: u16) {
self.port = new_port;
}
pub fn from_socket_addr(sa: &SocketAddr) -> Self {
match sa {
SocketAddr::V4(v) => Self {
addr: IpAddr::V4(*v.ip()),
port: v.port(),
},
SocketAddr::V6(v) => Self {
addr: IpAddr::V6(*v.ip()),
port: v.port(),
},
}
}
pub fn to_socket_addr(&self) -> SocketAddr {
match self.addr {
IpAddr::V4(a) => SocketAddr::V4(SocketAddrV4::new(a, self.port)),
IpAddr::V6(a) => SocketAddr::V6(SocketAddrV6::new(a, self.port, 0, 0)),
}
}
}
impl fmt::Display for IpAddrPort {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.addr {
IpAddr::V4(a) => write!(f, "{}:{}", a, self.port()),
IpAddr::V6(a) => write!(f, "[{}]:{}", a, self.port()),
}
}
}
impl From<SocketAddrV4> for IpAddrPort {
fn from(sock4: SocketAddrV4) -> IpAddrPort {
Self::from_socket_addr(&SocketAddr::V4(sock4))
}
}
impl From<SocketAddrV6> for IpAddrPort {
fn from(sock6: SocketAddrV6) -> IpAddrPort {
Self::from_socket_addr(&SocketAddr::V6(sock6))
}
}

268
veilid-tools/src/ip_extra.rs Normal file
View File

@@ -0,0 +1,268 @@
//
// This file really shouldn't be necessary, but 'ip' isn't a stable feature
//
use super::*;
use core::hash::*;
#[derive(Copy, PartialEq, Eq, Clone, Hash, Debug)]
pub enum Ipv6MulticastScope {
InterfaceLocal,
LinkLocal,
RealmLocal,
AdminLocal,
SiteLocal,
OrganizationLocal,
Global,
}
pub fn ipaddr_is_unspecified(addr: &IpAddr) -> bool {
match addr {
IpAddr::V4(ip) => ipv4addr_is_unspecified(ip),
IpAddr::V6(ip) => ipv6addr_is_unspecified(ip),
}
}
pub fn ipaddr_is_loopback(addr: &IpAddr) -> bool {
match addr {
IpAddr::V4(ip) => ipv4addr_is_loopback(ip),
IpAddr::V6(ip) => ipv6addr_is_loopback(ip),
}
}
pub fn ipaddr_is_global(addr: &IpAddr) -> bool {
match addr {
IpAddr::V4(ip) => ipv4addr_is_global(ip),
IpAddr::V6(ip) => ipv6addr_is_global(ip),
}
}
pub fn ipaddr_is_multicast(addr: &IpAddr) -> bool {
match addr {
IpAddr::V4(ip) => ipv4addr_is_multicast(ip),
IpAddr::V6(ip) => ipv6addr_is_multicast(ip),
}
}
pub fn ipaddr_is_documentation(addr: &IpAddr) -> bool {
match addr {
IpAddr::V4(ip) => ipv4addr_is_documentation(ip),
IpAddr::V6(ip) => ipv6addr_is_documentation(ip),
}
}
pub fn ipv4addr_is_unspecified(addr: &Ipv4Addr) -> bool {
addr.octets() == [0u8, 0u8, 0u8, 0u8]
}
pub fn ipv4addr_is_loopback(addr: &Ipv4Addr) -> bool {
addr.octets()[0] == 127
}
pub fn ipv4addr_is_private(addr: &Ipv4Addr) -> bool {
match addr.octets() {
[10, ..] => true,
[172, b, ..] if (16..=31).contains(&b) => true,
[192, 168, ..] => true,
_ => false,
}
}
pub fn ipv4addr_is_link_local(addr: &Ipv4Addr) -> bool {
matches!(addr.octets(), [169, 254, ..])
}
pub fn ipv4addr_is_global(addr: &Ipv4Addr) -> bool {
// check if this address is 192.0.0.9 or 192.0.0.10. These addresses are the only two
// globally routable addresses in the 192.0.0.0/24 range.
if u32::from(*addr) == 0xc0000009 || u32::from(*addr) == 0xc000000a {
return true;
}
!ipv4addr_is_private(addr)
&& !ipv4addr_is_loopback(addr)
&& !ipv4addr_is_link_local(addr)
&& !ipv4addr_is_broadcast(addr)
&& !ipv4addr_is_documentation(addr)
&& !ipv4addr_is_shared(addr)
&& !ipv4addr_is_ietf_protocol_assignment(addr)
&& !ipv4addr_is_reserved(addr)
&& !ipv4addr_is_benchmarking(addr)
// Make sure the address is not in 0.0.0.0/8
&& addr.octets()[0] != 0
}
pub fn ipv4addr_is_shared(addr: &Ipv4Addr) -> bool {
addr.octets()[0] == 100 && (addr.octets()[1] & 0b1100_0000 == 0b0100_0000)
}
pub fn ipv4addr_is_ietf_protocol_assignment(addr: &Ipv4Addr) -> bool {
addr.octets()[0] == 192 && addr.octets()[1] == 0 && addr.octets()[2] == 0
}
pub fn ipv4addr_is_benchmarking(addr: &Ipv4Addr) -> bool {
addr.octets()[0] == 198 && (addr.octets()[1] & 0xfe) == 18
}
pub fn ipv4addr_is_reserved(addr: &Ipv4Addr) -> bool {
addr.octets()[0] & 240 == 240 && !addr.is_broadcast()
}
pub fn ipv4addr_is_multicast(addr: &Ipv4Addr) -> bool {
addr.octets()[0] >= 224 && addr.octets()[0] <= 239
}
pub fn ipv4addr_is_broadcast(addr: &Ipv4Addr) -> bool {
addr.octets() == [255u8, 255u8, 255u8, 255u8]
}
pub fn ipv4addr_is_documentation(addr: &Ipv4Addr) -> bool {
matches!(
addr.octets(),
[192, 0, 2, _] | [198, 51, 100, _] | [203, 0, 113, _]
)
}
pub fn ipv6addr_is_unspecified(addr: &Ipv6Addr) -> bool {
addr.segments() == [0, 0, 0, 0, 0, 0, 0, 0]
}
pub fn ipv6addr_is_loopback(addr: &Ipv6Addr) -> bool {
addr.segments() == [0, 0, 0, 0, 0, 0, 0, 1]
}
pub fn ipv6addr_is_global(addr: &Ipv6Addr) -> bool {
match ipv6addr_multicast_scope(addr) {
Some(Ipv6MulticastScope::Global) => true,
None => ipv6addr_is_unicast_global(addr),
_ => false,
}
}
pub fn ipv6addr_is_unique_local(addr: &Ipv6Addr) -> bool {
(addr.segments()[0] & 0xfe00) == 0xfc00
}
pub fn ipv6addr_is_unicast_link_local_strict(addr: &Ipv6Addr) -> bool {
addr.segments()[0] == 0xfe80
&& addr.segments()[1] == 0
&& addr.segments()[2] == 0
&& addr.segments()[3] == 0
}
pub fn ipv6addr_is_unicast_link_local(addr: &Ipv6Addr) -> bool {
(addr.segments()[0] & 0xffc0) == 0xfe80
}
pub fn ipv6addr_is_unicast_site_local(addr: &Ipv6Addr) -> bool {
(addr.segments()[0] & 0xffc0) == 0xfec0
}
pub fn ipv6addr_is_documentation(addr: &Ipv6Addr) -> bool {
(addr.segments()[0] == 0x2001) && (addr.segments()[1] == 0xdb8)
}
pub fn ipv6addr_is_unicast_global(addr: &Ipv6Addr) -> bool {
!ipv6addr_is_multicast(addr)
&& !ipv6addr_is_loopback(addr)
&& !ipv6addr_is_unicast_link_local(addr)
&& !ipv6addr_is_unique_local(addr)
&& !ipv6addr_is_unspecified(addr)
&& !ipv6addr_is_documentation(addr)
}
pub fn ipv6addr_multicast_scope(addr: &Ipv6Addr) -> Option<Ipv6MulticastScope> {
if ipv6addr_is_multicast(addr) {
match addr.segments()[0] & 0x000f {
1 => Some(Ipv6MulticastScope::InterfaceLocal),
2 => Some(Ipv6MulticastScope::LinkLocal),
3 => Some(Ipv6MulticastScope::RealmLocal),
4 => Some(Ipv6MulticastScope::AdminLocal),
5 => Some(Ipv6MulticastScope::SiteLocal),
8 => Some(Ipv6MulticastScope::OrganizationLocal),
14 => Some(Ipv6MulticastScope::Global),
_ => None,
}
} else {
None
}
}
pub fn ipv6addr_is_multicast(addr: &Ipv6Addr) -> bool {
(addr.segments()[0] & 0xff00) == 0xff00
}
// Converts an ip to a ip block by applying a netmask
// to the host part of the ip address
// ipv4 addresses are treated as single hosts
// ipv6 addresses are treated as prefix allocated blocks
pub fn ip_to_ipblock(ip6_prefix_size: usize, addr: IpAddr) -> IpAddr {
match addr {
IpAddr::V4(_) => addr,
IpAddr::V6(v6) => {
let mut hostlen = 128usize.saturating_sub(ip6_prefix_size);
let mut out = v6.octets();
for i in (0..16).rev() {
if hostlen >= 8 {
out[i] = 0xFF;
hostlen -= 8;
} else {
out[i] |= !(0xFFu8 << hostlen);
break;
}
}
IpAddr::V6(Ipv6Addr::from(out))
}
}
}
pub fn ipaddr_apply_netmask(addr: IpAddr, netmask: IpAddr) -> IpAddr {
match addr {
IpAddr::V4(v4) => {
let v4mask = match netmask {
IpAddr::V4(v4mask) => v4mask,
IpAddr::V6(_) => {
panic!("netmask doesn't match ipv4 address");
}
};
let v4 = v4.octets();
let v4mask = v4mask.octets();
IpAddr::V4(Ipv4Addr::new(
v4[0] & v4mask[0],
v4[1] & v4mask[1],
v4[2] & v4mask[2],
v4[3] & v4mask[3],
))
}
IpAddr::V6(v6) => {
let v6mask = match netmask {
IpAddr::V4(_) => {
panic!("netmask doesn't match ipv6 address");
}
IpAddr::V6(v6mask) => v6mask,
};
let v6 = v6.segments();
let v6mask = v6mask.segments();
IpAddr::V6(Ipv6Addr::new(
v6[0] & v6mask[0],
v6[1] & v6mask[1],
v6[2] & v6mask[2],
v6[3] & v6mask[3],
v6[4] & v6mask[4],
v6[5] & v6mask[5],
v6[6] & v6mask[6],
v6[7] & v6mask[7],
))
}
}
}
pub fn ipaddr_in_network(addr: IpAddr, netaddr: IpAddr, netmask: IpAddr) -> bool {
if addr.is_ipv4() && !netaddr.is_ipv4() {
return false;
}
if addr.is_ipv6() && !netaddr.is_ipv6() {
return false;
}
ipaddr_apply_netmask(netaddr, netmask) == ipaddr_apply_netmask(addr, netmask)
}

353
veilid-tools/src/log_thru.rs Normal file
View File

@@ -0,0 +1,353 @@
// LogThru
// Pass errors through and log them simultaneously via map_err()
// Also contains common log facilities (net, rpc, rtab, pstore, crypto, etc )
use alloc::string::{String, ToString};
pub fn map_to_string<X: ToString>(arg: X) -> String {
arg.to_string()
}
#[macro_export]
macro_rules! fn_string {
($text:expr) => {
|| $text.to_string()
};
}
#[macro_export]
macro_rules! log_net {
(error $text:expr) => {error!(
target: "net",
"{}",
$text,
)};
(error $fmt:literal, $($arg:expr),+) => {
error!(target:"net", $fmt, $($arg),+);
};
(warn $text:expr) => {warn!(
target: "net",
"{}",
$text,
)};
(warn $fmt:literal, $($arg:expr),+) => {
warn!(target:"net", $fmt, $($arg),+);
};
(debug $text:expr) => {debug!(
target: "net",
"{}",
$text,
)};
(debug $fmt:literal, $($arg:expr),+) => {
debug!(target:"net", $fmt, $($arg),+);
};
($text:expr) => {trace!(
target: "net",
"{}",
$text,
)};
($fmt:literal, $($arg:expr),+) => {
trace!(target:"net", $fmt, $($arg),+);
}
}
#[macro_export]
macro_rules! log_rpc {
(error $text:expr) => { error!(
target: "rpc",
"{}",
$text,
)};
(error $fmt:literal, $($arg:expr),+) => {
error!(target:"rpc", $fmt, $($arg),+);
};
(warn $text:expr) => { warn!(
target: "rpc",
"{}",
$text,
)};
(warn $fmt:literal, $($arg:expr),+) => {
warn!(target:"rpc", $fmt, $($arg),+);
};
(debug $text:expr) => { error!(
target: "rpc",
"{}",
$text,
)};
(debug $fmt:literal, $($arg:expr),+) => {
debug!(target:"rpc", $fmt, $($arg),+);
};
($text:expr) => {trace!(
target: "rpc",
"{}",
$text,
)};
($fmt:literal, $($arg:expr),+) => {
trace!(target:"rpc", $fmt, $($arg),+);
}
}
#[macro_export]
macro_rules! log_rtab {
(error $text:expr) => { error!(
target: "rtab",
"{}",
$text,
)};
(error $fmt:literal, $($arg:expr),+) => {
error!(target:"rtab", $fmt, $($arg),+);
};
(warn $text:expr) => { warn!(
target: "rtab",
"{}",
$text,
)};
(warn $fmt:literal, $($arg:expr),+) => {
warn!(target:"rtab", $fmt, $($arg),+);
};
(debug $text:expr) => { debug!(
target: "rtab",
"{}",
$text,
)};
(debug $fmt:literal, $($arg:expr),+) => {
debug!(target:"rtab", $fmt, $($arg),+);
};
($text:expr) => {trace!(
target: "rtab",
"{}",
$text,
)};
($fmt:literal, $($arg:expr),+) => {
trace!(target:"rtab", $fmt, $($arg),+);
}
}
#[macro_export]
macro_rules! log_pstore {
(error $text:expr) => { error!(
target: "pstore",
"{}",
$text,
)};
(error $fmt:literal, $($arg:expr),+) => {
error!(target:"pstore", $fmt, $($arg),+);
};
(warn $text:expr) => { warn!(
target: "pstore",
"{}",
$text,
)};
(warn $fmt:literal, $($arg:expr),+) => {
warn!(target:"pstore", $fmt, $($arg),+);
};
($text:expr) => {trace!(
target: "pstore",
"{}",
$text,
)};
($fmt:literal, $($arg:expr),+) => {
trace!(target:"pstore", $fmt, $($arg),+);
}
}
#[macro_export]
macro_rules! log_crypto {
(error $text:expr) => { error!(
target: "crypto",
"{}",
$text,
)};
(error $fmt:literal, $($arg:expr),+) => {
error!(target:"crypto", $fmt, $($arg),+);
};
(warn $text:expr) => { warn!(
target: "crypto",
"{}",
$text,
)};
(warn $fmt:literal, $($arg:expr),+) => {
warn!(target:"crypto", $fmt, $($arg),+);
};
($text:expr) => {trace!(
target: "crypto",
"{}",
$text,
)};
($fmt:literal, $($arg:expr),+) => {
trace!(target:"crypto", $fmt, $($arg),+);
}
}
#[macro_export]
macro_rules! logthru_net {
($($level:ident)?) => {
logthru!($($level)? "net")
};
($($level:ident)? $text:literal) => {
logthru!($($level)? "net", $text)
};
($($level:ident)? $fmt:literal, $($arg:expr),+) => {
logthru!($($level)? "net", $fmt, $($arg),+)
}
}
#[macro_export]
macro_rules! logthru_rpc {
($($level:ident)?) => {
logthru!($($level)? "rpc")
};
($($level:ident)? $text:literal) => {
logthru!($($level)? "rpc", $text)
};
($($level:ident)? $fmt:literal, $($arg:expr),+) => {
logthru!($($level)? "rpc", $fmt, $($arg),+)
}
}
#[macro_export]
macro_rules! logthru_rtab {
($($level:ident)?) => {
logthru!($($level)? "rtab")
};
($($level:ident)? $text:literal) => {
logthru!($($level)? "rtab", $text)
};
($($level:ident)? $fmt:literal, $($arg:expr),+) => {
logthru!($($level)? "rtab", $fmt, $($arg),+)
}
}
#[macro_export]
macro_rules! logthru_pstore {
($($level:ident)?) => {
logthru!($($level)? "pstore")
};
($($level:ident)? $text:literal) => {
logthru!($($level)? "pstore", $text)
};
($($level:ident)? $fmt:literal, $($arg:expr),+) => {
logthru!($($level)? "pstore", $fmt, $($arg),+)
}
}
#[macro_export]
macro_rules! logthru_crypto {
($($level:ident)?) => {
logthru!($($level)? "crypto")
};
($($level:ident)? $text:literal) => {
logthru!($($level)? "crypto", $text)
};
($($level:ident)? $fmt:literal, $($arg:expr),+) => {
logthru!($($level)? "crypto", $fmt, $($arg),+)
}
}
#[macro_export]
macro_rules! logthru {
// error
(error $target:literal) => (|e__| {
error!(
target: $target,
"[{:?}]",
e__,
);
e__
});
(error $target:literal, $text:literal) => (|e__| {
error!(
target: $target,
"[{:?}] {}",
e__,
$text
);
e__
});
(error $target:literal, $fmt:literal, $($arg:expr),+) => (|e__| {
error!(
target: $target,
concat!("[{:?}] ", $fmt),
e__,
$($arg),+
);
e__
});
// warn
(warn $target:literal) => (|e__| {
warn!(
target: $target,
"[{:?}]",
e__,
);
e__
});
(warn $target:literal, $text:literal) => (|e__| {
warn!(
target: $target,
"[{:?}] {}",
e__,
$text
);
e__
});
(warn $target:literal, $fmt:literal, $($arg:expr),+) => (|e__| {
warn!(
target: $target,
concat!("[{:?}] ", $fmt),
e__,
$($arg),+
);
e__
});
// debug
(debug $target:literal) => (|e__| {
debug!(
target: $target,
"[{:?}]",
e__,
);
e__
});
(debug $target:literal, $text:literal) => (|e__| {
debug!(
target: $target,
"[{:?}] {}",
e__,
$text
);
e__
});
(debug $target:literal, $fmt:literal, $($arg:expr),+) => (|e__| {
debug!(
target: $target,
concat!("[{:?}] ", $fmt),
e__,
$($arg),+
);
e__
});
// trace
($target:literal) => (|e__| {
trace!(
target: $target,
"[{:?}]",
e__,
);
e__
});
($target:literal, $text:literal) => (|e__| {
trace!(
target: $target,
"[{:?}] {}",
e__,
$text
);
e__
});
($target:literal, $fmt:literal, $($arg:expr),+) => (|e__| {
trace!(
target: $target,
concat!("[{:?}] ", $fmt),
e__,
$($arg),+
);
e__
})
}

134
veilid-tools/src/mod.rs Normal file
View File

@@ -0,0 +1,134 @@
// mod bump_port;
mod async_peek_stream;
mod async_tag_lock;
mod clone_stream;
mod eventual;
mod eventual_base;
mod eventual_value;
mod eventual_value_clone;
mod interval;
mod ip_addr_port;
mod ip_extra;
mod log_thru;
mod must_join_handle;
mod must_join_single_future;
mod mutable_future;
mod network_result;
mod random;
mod single_shot_eventual;
mod sleep;
mod spawn;
mod split_url;
mod tick_task;
mod timeout;
mod timeout_or;
mod timestamp;
mod tools;
#[cfg(target_arch = "wasm32")]
mod wasm;
pub use cfg_if::*;
#[allow(unused_imports)]
pub use eyre::{bail, ensure, eyre, Report as EyreReport, Result as EyreResult, WrapErr};
pub use futures_util::future::{select, Either};
pub use futures_util::select;
pub use futures_util::stream::FuturesUnordered;
pub use futures_util::{AsyncRead, AsyncWrite};
pub use log_thru::*;
pub use owo_colors::OwoColorize;
pub use parking_lot::*;
pub use split_url::*;
pub use static_assertions::*;
pub use stop_token::*;
pub use thiserror::Error as ThisError;
cfg_if! {
if #[cfg(feature = "tracing")] {
pub use tracing::*;
} else {
pub use log::*;
}
}
pub type PinBox<T> = Pin<Box<T>>;
pub type PinBoxFuture<T> = PinBox<dyn Future<Output = T> + 'static>;
pub type PinBoxFutureLifetime<'a, T> = PinBox<dyn Future<Output = T> + 'a>;
pub type SendPinBoxFuture<T> = PinBox<dyn Future<Output = T> + Send + 'static>;
pub type SendPinBoxFutureLifetime<'a, T> = PinBox<dyn Future<Output = T> + Send + 'a>;
pub use std::borrow::{Cow, ToOwned};
pub use std::boxed::Box;
pub use std::cell::RefCell;
pub use std::cmp;
pub use std::collections::btree_map::BTreeMap;
pub use std::collections::btree_set::BTreeSet;
pub use std::collections::hash_map::HashMap;
pub use std::collections::hash_set::HashSet;
pub use std::collections::LinkedList;
pub use std::collections::VecDeque;
pub use std::convert::{TryFrom, TryInto};
pub use std::fmt;
pub use std::future::Future;
pub use std::mem;
pub use std::net::{
IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs,
};
pub use std::ops::{Fn, FnMut, FnOnce};
pub use std::pin::Pin;
pub use std::rc::Rc;
pub use std::string::String;
pub use std::sync::atomic::{AtomicBool, Ordering};
pub use std::sync::{Arc, Weak};
pub use std::task;
pub use std::time::Duration;
pub use std::vec::Vec;
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
pub use async_lock::Mutex as AsyncMutex;
pub use async_lock::MutexGuard as AsyncMutexGuard;
pub use async_lock::MutexGuardArc as AsyncMutexGuardArc;
pub use async_executors::JoinHandle as LowLevelJoinHandle;
} else {
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::sync::MutexGuardArc as AsyncMutexGuardArc;
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::sync::OwnedMutexGuard as AsyncMutexGuardArc;
pub use tokio::task::JoinHandle as LowLevelJoinHandle;
} else {
#[compile_error("must use an executor")]
}
}
}
}
// pub use bump_port::*;
pub use async_peek_stream::*;
pub use async_tag_lock::*;
pub use clone_stream::*;
pub use eventual::*;
pub use eventual_base::{EventualCommon, EventualResolvedFuture};
pub use eventual_value::*;
pub use eventual_value_clone::*;
pub use interval::*;
pub use ip_addr_port::*;
pub use ip_extra::*;
pub use must_join_handle::*;
pub use must_join_single_future::*;
pub use mutable_future::*;
pub use network_result::*;
pub use random::*;
pub use single_shot_eventual::*;
pub use sleep::*;
pub use spawn::*;
pub use tick_task::*;
pub use timeout::*;
pub use timeout_or::*;
pub use timestamp::*;
pub use tools::*;
#[cfg(target_arch = "wasm32")]
pub use wasm::*;

90
veilid-tools/src/must_join_handle.rs Normal file
View File

@@ -0,0 +1,90 @@
use super::*;
use core::task::{Context, Poll};
#[derive(Debug)]
pub struct MustJoinHandle<T> {
join_handle: Option<LowLevelJoinHandle<T>>,
completed: bool,
}
impl<T> MustJoinHandle<T> {
pub fn new(join_handle: LowLevelJoinHandle<T>) -> Self {
Self {
join_handle: Some(join_handle),
completed: false,
}
}
#[allow(unused_mut)]
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;
}
} else if #[cfg(target_arch = "wasm32")] {
drop(self.join_handle.take());
self.completed = true;
} else {
compile_error!("needs executor implementation")
}
}
}
}
}
impl<T> Drop for MustJoinHandle<T> {
fn drop(&mut self) {
// panic if we haven't completed
if !self.completed {
panic!("MustJoinHandle was not completed upon drop. Add cooperative cancellation where appropriate to ensure this is completed before drop.")
}
}
}
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(self.join_handle.as_mut().unwrap()).poll(cx) {
Poll::Ready(t) => {
if self.completed {
panic!("should not poll completed join handle");
}
self.completed = true;
cfg_if! {
if #[cfg(feature="rt-async-std")] {
Poll::Ready(t)
} else if #[cfg(feature="rt-tokio")] {
match t {
Ok(t) => Poll::Ready(t),
Err(e) => {
if e.is_panic() {
// Resume the panic on the main task
std::panic::resume_unwind(e.into_panic());
} else {
panic!("join error was not a panic, should not poll after abort");
}
}
}
}else if #[cfg(target_arch = "wasm32")] {
Poll::Ready(t)
} else {
compile_error!("needs executor implementation")
}
}
}
Poll::Pending => Poll::Pending,
}
}
}

205
veilid-tools/src/must_join_single_future.rs Normal file
View File

@@ -0,0 +1,205 @@
use super::*;
use core::task::Poll;
use futures_util::poll;
#[derive(Debug)]
struct MustJoinSingleFutureInner<T>
where
T: 'static,
{
locked: bool,
join_handle: Option<MustJoinHandle<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 MustJoinSingleFuture<T>
where
T: 'static,
{
inner: Arc<Mutex<MustJoinSingleFutureInner<T>>>,
}
impl<T> Default for MustJoinSingleFuture<T>
where
T: 'static,
{
fn default() -> Self {
Self::new()
}
}
impl<T> MustJoinSingleFuture<T>
where
T: 'static,
{
pub fn new() -> Self {
Self {
inner: Arc::new(Mutex::new(MustJoinSingleFutureInner {
locked: false,
join_handle: None,
})),
}
}
fn try_lock(&self) -> Result<Option<MustJoinHandle<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<MustJoinHandle<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 and take it if there is one
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 and take it
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)
}
// Possibly spawn the future possibly returning the value of the last execution
pub async fn single_spawn_local(
&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))
}
}
impl<T> MustJoinSingleFuture<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))
}
}

33
veilid-tools/src/mutable_future.rs Normal file
View File

@@ -0,0 +1,33 @@
use super::*;
pub struct MutableFuture<O, T: Future<Output = O>> {
inner: Arc<Mutex<Pin<Box<T>>>>,
}
impl<O, T: Future<Output = O>> MutableFuture<O, T> {
pub fn new(inner: T) -> Self {
Self {
inner: Arc::new(Mutex::new(Box::pin(inner))),
}
}
pub fn set(&self, inner: T) {
*self.inner.lock() = Box::pin(inner);
}
}
impl<O, T: Future<Output = O>> Clone for MutableFuture<O, T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<O, T: Future<Output = O>> Future for MutableFuture<O, T> {
type Output = O;
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let mut inner = self.inner.lock();
T::poll(inner.as_mut(), cx)
}
}

379
veilid-tools/src/network_result.rs Normal file
View File

@@ -0,0 +1,379 @@
use super::*;
use core::fmt::{Debug, Display};
use core::result::Result;
use std::error::Error;
use std::io;
//////////////////////////////////////////////////////////////////
// Non-fallible network results conversions
pub trait NetworkResultExt<T> {
fn into_network_result(self) -> NetworkResult<T>;
}
impl<T> NetworkResultExt<T> for Result<T, TimeoutError> {
fn into_network_result(self) -> NetworkResult<T> {
self.ok()
.map(|v| NetworkResult::<T>::Value(v))
.unwrap_or(NetworkResult::<T>::Timeout)
}
}
pub trait IoNetworkResultExt<T> {
fn into_network_result(self) -> io::Result<NetworkResult<T>>;
}
impl<T> IoNetworkResultExt<T> for io::Result<T> {
fn into_network_result(self) -> io::Result<NetworkResult<T>> {
match self {
Ok(v) => Ok(NetworkResult::Value(v)),
#[cfg(feature = "io_error_more")]
Err(e) => match e.kind() {
io::ErrorKind::TimedOut => Ok(NetworkResult::Timeout),
io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionRefused
| io::ErrorKind::ConnectionReset
| io::ErrorKind::HostUnreachable
| io::ErrorKind::NetworkUnreachable => Ok(NetworkResult::NoConnection(e)),
io::ErrorKind::AddrNotAvailable => Ok(NetworkResult::AlreadyExists(e)),
_ => Err(e),
},
#[cfg(not(feature = "io_error_more"))]
Err(e) => {
#[cfg(not(target_arch = "wasm32"))]
if let Some(os_err) = e.raw_os_error() {
if os_err == libc::EHOSTUNREACH || os_err == libc::ENETUNREACH {
return Ok(NetworkResult::NoConnection(e));
}
}
match e.kind() {
io::ErrorKind::TimedOut => Ok(NetworkResult::Timeout),
io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionRefused
| io::ErrorKind::ConnectionReset => Ok(NetworkResult::NoConnection(e)),
io::ErrorKind::AddrNotAvailable => Ok(NetworkResult::AlreadyExists(e)),
_ => Err(e),
}
}
}
}
}
pub trait NetworkResultResultExt<T, E> {
fn into_result_network_result(self) -> Result<NetworkResult<T>, E>;
}
impl<T, E> NetworkResultResultExt<T, E> for NetworkResult<Result<T, E>> {
fn into_result_network_result(self) -> Result<NetworkResult<T>, E> {
match self {
NetworkResult::Timeout => Ok(NetworkResult::<T>::Timeout),
NetworkResult::NoConnection(e) => Ok(NetworkResult::<T>::NoConnection(e)),
NetworkResult::AlreadyExists(e) => Ok(NetworkResult::<T>::AlreadyExists(e)),
NetworkResult::InvalidMessage(s) => Ok(NetworkResult::<T>::InvalidMessage(s)),
NetworkResult::Value(Ok(v)) => Ok(NetworkResult::<T>::Value(v)),
NetworkResult::Value(Err(e)) => Err(e),
}
}
}
pub trait FoldedNetworkResultExt<T> {
fn folded(self) -> io::Result<NetworkResult<T>>;
}
impl<T> FoldedNetworkResultExt<T> for io::Result<TimeoutOr<T>> {
fn folded(self) -> io::Result<NetworkResult<T>> {
match self {
Ok(TimeoutOr::Timeout) => Ok(NetworkResult::Timeout),
Ok(TimeoutOr::Value(v)) => Ok(NetworkResult::Value(v)),
#[cfg(feature = "io_error_more")]
Err(e) => match e.kind() {
io::ErrorKind::TimedOut => Ok(NetworkResult::Timeout),
io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionRefused
| io::ErrorKind::ConnectionReset
| io::ErrorKind::HostUnreachable
| io::ErrorKind::NetworkUnreachable => Ok(NetworkResult::NoConnection(e)),
io::ErrorKind::AddrNotAvailable => Ok(NetworkResult::AlreadyExists(e)),
_ => Err(e),
},
#[cfg(not(feature = "io_error_more"))]
Err(e) => {
#[cfg(not(target_arch = "wasm32"))]
if let Some(os_err) = e.raw_os_error() {
if os_err == libc::EHOSTUNREACH || os_err == libc::ENETUNREACH {
return Ok(NetworkResult::NoConnection(e));
}
}
match e.kind() {
io::ErrorKind::TimedOut => Ok(NetworkResult::Timeout),
io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionRefused
| io::ErrorKind::ConnectionReset => Ok(NetworkResult::NoConnection(e)),
io::ErrorKind::AddrNotAvailable => Ok(NetworkResult::AlreadyExists(e)),
_ => Err(e),
}
}
}
}
}
impl<T> FoldedNetworkResultExt<T> for io::Result<NetworkResult<T>> {
fn folded(self) -> io::Result<NetworkResult<T>> {
match self {
Ok(v) => Ok(v),
#[cfg(feature = "io_error_more")]
Err(e) => match e.kind() {
io::ErrorKind::TimedOut => Ok(NetworkResult::Timeout),
io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionRefused
| io::ErrorKind::ConnectionReset
| io::ErrorKind::HostUnreachable
| io::ErrorKind::NetworkUnreachable => Ok(NetworkResult::NoConnection(e)),
io::ErrorKind::AddrNotAvailable => Ok(NetworkResult::AlreadyExists(e)),
_ => Err(e),
},
#[cfg(not(feature = "io_error_more"))]
Err(e) => {
#[cfg(not(target_arch = "wasm32"))]
if let Some(os_err) = e.raw_os_error() {
if os_err == libc::EHOSTUNREACH || os_err == libc::ENETUNREACH {
return Ok(NetworkResult::NoConnection(e));
}
}
match e.kind() {
io::ErrorKind::TimedOut => Ok(NetworkResult::Timeout),
io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionRefused
| io::ErrorKind::ConnectionReset => Ok(NetworkResult::NoConnection(e)),
io::ErrorKind::AddrNotAvailable => Ok(NetworkResult::AlreadyExists(e)),
_ => Err(e),
}
}
}
}
}
//////////////////////////////////////////////////////////////////
// Non-fallible network result
#[must_use]
pub enum NetworkResult<T> {
Timeout,
NoConnection(io::Error),
AlreadyExists(io::Error),
InvalidMessage(String),
Value(T),
}
impl<T> NetworkResult<T> {
pub fn timeout() -> Self {
Self::Timeout
}
pub fn no_connection(e: io::Error) -> Self {
Self::NoConnection(e)
}
pub fn no_connection_other<S: ToString>(s: S) -> Self {
Self::NoConnection(io::Error::new(io::ErrorKind::Other, s.to_string()))
}
pub fn invalid_message<S: ToString>(s: S) -> Self {
Self::InvalidMessage(s.to_string())
}
pub fn already_exists(e: io::Error) -> Self {
Self::AlreadyExists(e)
}
pub fn value(value: T) -> Self {
Self::Value(value)
}
pub fn is_timeout(&self) -> bool {
matches!(self, Self::Timeout)
}
pub fn is_no_connection(&self) -> bool {
matches!(self, Self::NoConnection(_))
}
pub fn is_already_exists(&self) -> bool {
matches!(self, Self::AlreadyExists(_))
}
pub fn is_value(&self) -> bool {
matches!(self, Self::Value(_))
}
pub fn map<X, F: Fn(T) -> X>(self, f: F) -> NetworkResult<X> {
match self {
Self::Timeout => NetworkResult::<X>::Timeout,
Self::NoConnection(e) => NetworkResult::<X>::NoConnection(e),
Self::AlreadyExists(e) => NetworkResult::<X>::AlreadyExists(e),
Self::InvalidMessage(s) => NetworkResult::<X>::InvalidMessage(s),
Self::Value(v) => NetworkResult::<X>::Value(f(v)),
}
}
pub fn into_result(self) -> Result<T, io::Error> {
match self {
Self::Timeout => Err(io::Error::new(io::ErrorKind::TimedOut, "Timed out")),
Self::NoConnection(e) => Err(e),
Self::AlreadyExists(e) => Err(e),
Self::InvalidMessage(s) => Err(io::Error::new(
io::ErrorKind::InvalidData,
format!("Invalid message: {}", s),
)),
Self::Value(v) => Ok(v),
}
}
}
impl<T> From<NetworkResult<T>> for Option<T> {
fn from(t: NetworkResult<T>) -> Self {
match t {
NetworkResult::Value(v) => Some(v),
_ => None,
}
}
}
// impl<T: Clone> Clone for NetworkResult<T> {
// fn clone(&self) -> Self {
// match self {
// Self::Timeout => Self::Timeout,
// Self::NoConnection(e) => Self::NoConnection(e.clone()),
// Self::InvalidMessage(s) => Self::InvalidMessage(s.clone()),
// Self::Value(t) => Self::Value(t.clone()),
// }
// }
// }
impl<T: Debug> Debug for NetworkResult<T> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Timeout => write!(f, "Timeout"),
Self::NoConnection(e) => f.debug_tuple("NoConnection").field(e).finish(),
Self::AlreadyExists(e) => f.debug_tuple("AlreadyExists").field(e).finish(),
Self::InvalidMessage(s) => f.debug_tuple("InvalidMessage").field(s).finish(),
Self::Value(v) => f.debug_tuple("Value").field(v).finish(),
}
}
}
impl<T> Display for NetworkResult<T> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Timeout => write!(f, "Timeout"),
Self::NoConnection(e) => write!(f, "NoConnection({})", e.kind()),
Self::AlreadyExists(e) => write!(f, "AlreadyExists({})", e.kind()),
Self::InvalidMessage(s) => write!(f, "InvalidMessage({})", s),
Self::Value(_) => write!(f, "Value"),
}
}
}
impl<T: Debug + Display> Error for NetworkResult<T> {}
//////////////////////////////////////////////////////////////////
// Non-fallible network result macros
#[macro_export]
macro_rules! network_result_try {
($r: expr) => {
match $r {
NetworkResult::Timeout => return Ok(NetworkResult::Timeout),
NetworkResult::NoConnection(e) => return Ok(NetworkResult::NoConnection(e)),
NetworkResult::AlreadyExists(e) => return Ok(NetworkResult::AlreadyExists(e)),
NetworkResult::InvalidMessage(s) => return Ok(NetworkResult::InvalidMessage(s)),
NetworkResult::Value(v) => v,
}
};
($r:expr => $f:tt) => {
match $r {
NetworkResult::Timeout => {
$f;
return Ok(NetworkResult::Timeout);
}
NetworkResult::NoConnection(e) => {
$f;
return Ok(NetworkResult::NoConnection(e));
}
NetworkResult::AlreadyExists(e) => {
$f;
return Ok(NetworkResult::AlreadyExists(e));
}
NetworkResult::InvalidMessage(s) => {
$f;
return Ok(NetworkResult::InvalidMessage(s));
}
NetworkResult::Value(v) => v,
}
};
}
#[macro_export]
macro_rules! log_network_result {
($text:expr) => {
cfg_if::cfg_if! {
if #[cfg(debug_assertions)] {
info!(target: "network_result", "{}", $text)
} else {
debug!(target: "network_result", "{}", $text)
}
}
};
($fmt:literal, $($arg:expr),+) => {
cfg_if::cfg_if! {
if #[cfg(debug_assertions)] {
info!(target: "network_result", $fmt, $($arg),+);
} else {
debug!(target: "network_result", $fmt, $($arg),+);
}
}
};
}
#[macro_export]
macro_rules! network_result_value_or_log {
($level: ident $r: expr => $f:tt) => {
match $r {
NetworkResult::Timeout => {
log_network_result!(
"{} at {}@{}:{}",
"Timeout".cyan(),
file!(),
line!(),
column!()
);
$f
}
NetworkResult::NoConnection(e) => {
log_network_result!(
"{}({}) at {}@{}:{}",
"No connection".cyan(),
e.to_string(),
file!(),
line!(),
column!()
);
$f
}
NetworkResult::AlreadyExists(e) => {
log_network_result!(
"{}({}) at {}@{}:{}",
"Already exists".cyan(),
e.to_string(),
file!(),
line!(),
column!()
);
$f
}
NetworkResult::InvalidMessage(s) => {
log_network_result!(
"{}({}) at {}@{}:{}",
"Invalid message".cyan(),
s,
file!(),
line!(),
column!()
);
$f
}
NetworkResult::Value(v) => v,
}
};
}

81
veilid-tools/src/random.rs Normal file
View File

@@ -0,0 +1,81 @@
use super::*;
use rand::prelude::*;
#[derive(Clone, Copy, Debug, Default)]
pub struct VeilidRng;
impl CryptoRng for VeilidRng {}
impl RngCore for VeilidRng {
fn next_u32(&mut self) -> u32 {
get_random_u32()
}
fn next_u64(&mut self) -> u64 {
get_random_u64()
}
fn fill_bytes(&mut self, dest: &mut [u8]) {
if let Err(e) = self.try_fill_bytes(dest) {
panic!("Error: {}", e);
}
}
fn try_fill_bytes(&mut self, dest: &mut [u8]) -> Result<(), rand::Error> {
random_bytes(dest).map_err(rand::Error::new)
}
}
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
pub fn random_bytes(dest: &mut [u8]) -> EyreResult<()> {
let len = dest.len();
let u32len = len / 4;
let remlen = len % 4;
for n in 0..u32len {
let r = (Math::random() * (u32::max_value() as f64)) as u32;
dest[n * 4 + 0] = (r & 0xFF) as u8;
dest[n * 4 + 1] = ((r >> 8) & 0xFF) as u8;
dest[n * 4 + 2] = ((r >> 16) & 0xFF) as u8;
dest[n * 4 + 3] = ((r >> 24) & 0xFF) as u8;
}
if remlen > 0 {
let r = (Math::random() * (u32::max_value() as f64)) as u32;
for n in 0..remlen {
dest[u32len * 4 + n] = ((r >> (n * 8)) & 0xFF) as u8;
}
}
Ok(())
}
pub fn get_random_u32() -> u32 {
(Math::random() * (u32::max_value() as f64)) as u32
}
pub fn get_random_u64() -> u64 {
let v1: u32 = get_random_u32();
let v2: u32 = get_random_u32();
((v1 as u64) << 32) | ((v2 as u32) as u64)
}
} else {
pub fn random_bytes(dest: &mut [u8]) -> EyreResult<()> {
let mut rng = rand::thread_rng();
rng.try_fill_bytes(dest).wrap_err("failed to fill bytes")
}
pub fn get_random_u32() -> u32 {
let mut rng = rand::thread_rng();
rng.next_u32()
}
pub fn get_random_u64() -> u64 {
let mut rng = rand::thread_rng();
rng.next_u64()
}
}
}

44
veilid-tools/src/single_shot_eventual.rs Normal file
View File

@@ -0,0 +1,44 @@
use super::*;
pub struct SingleShotEventual<T>
where
T: Unpin,
{
eventual: EventualValue<T>,
drop_value: Option<T>,
}
impl<T> Drop for SingleShotEventual<T>
where
T: Unpin,
{
fn drop(&mut self) {
if let Some(drop_value) = self.drop_value.take() {
self.eventual.resolve(drop_value);
}
}
}
impl<T> SingleShotEventual<T>
where
T: Unpin,
{
pub fn new(drop_value: Option<T>) -> Self {
Self {
eventual: EventualValue::new(),
drop_value,
}
}
// Can only call this once, it consumes the eventual
pub fn resolve(mut self, value: T) -> EventualResolvedFuture<EventualValue<T>> {
// If we resolve, we don't want to resolve again to the drop value
self.drop_value = None;
// Resolve to the specified value
self.eventual.resolve(value)
}
pub fn instance(&self) -> EventualValueFuture<T> {
self.eventual.instance()
}
}

34
veilid-tools/src/sleep.rs Normal file
View File

@@ -0,0 +1,34 @@
use super::*;
use std::time::Duration;
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
use async_executors::Bindgen;
pub async fn sleep(millis: u32) {
Bindgen.sleep(Duration::from_millis(millis.into())).await
}
} else {
pub async fn sleep(millis: u32) {
if millis == 0 {
cfg_if! {
if #[cfg(feature="rt-async-std")] {
async_std::task::yield_now().await;
} else if #[cfg(feature="rt-tokio")] {
tokio::task::yield_now().await;
}
}
} else {
cfg_if! {
if #[cfg(feature="rt-async-std")] {
async_std::task::sleep(Duration::from_millis(u64::from(millis))).await;
} else if #[cfg(feature="rt-tokio")] {
tokio::time::sleep(Duration::from_millis(u64::from(millis))).await;
}
}
}
}
}
}

119
veilid-tools/src/spawn.rs Normal file
View File

@@ -0,0 +1,119 @@
use super::*;
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
use async_executors::{Bindgen, LocalSpawnHandleExt, SpawnHandleExt};
pub fn spawn<Out>(future: impl Future<Output = Out> + Send + 'static) -> MustJoinHandle<Out>
where
Out: Send + 'static,
{
MustJoinHandle::new(
Bindgen
.spawn_handle(future)
.expect("wasm-bindgen-futures spawn_handle_local should never error out"),
)
}
pub fn spawn_local<Out>(future: impl Future<Output = Out> + 'static) -> MustJoinHandle<Out>
where
Out: 'static,
{
MustJoinHandle::new(
Bindgen
.spawn_handle_local(future)
.expect("wasm-bindgen-futures spawn_handle_local should never error out"),
)
}
pub fn spawn_detached<Out>(future: impl Future<Output = Out> + Send + 'static)
where
Out: Send + 'static,
{
Bindgen
.spawn_handle_local(future)
.expect("wasm-bindgen-futures spawn_handle_local should never error out")
.detach()
}
pub fn spawn_detached_local<Out>(future: impl Future<Output = Out> + 'static)
where
Out: 'static,
{
Bindgen
.spawn_handle_local(future)
.expect("wasm-bindgen-futures spawn_handle_local should never error out")
.detach()
}
} else {
pub fn spawn<Out>(future: impl Future<Output = Out> + Send + 'static) -> MustJoinHandle<Out>
where
Out: Send + 'static,
{
cfg_if! {
if #[cfg(feature="rt-async-std")] {
MustJoinHandle::new(async_std::task::spawn(future))
} else if #[cfg(feature="rt-tokio")] {
MustJoinHandle::new(tokio::task::spawn(future))
}
}
}
pub fn spawn_local<Out>(future: impl Future<Output = Out> + 'static) -> MustJoinHandle<Out>
where
Out: 'static,
{
cfg_if! {
if #[cfg(feature="rt-async-std")] {
MustJoinHandle::new(async_std::task::spawn_local(future))
} else if #[cfg(feature="rt-tokio")] {
MustJoinHandle::new(tokio::task::spawn_local(future))
}
}
}
pub fn spawn_detached<Out>(future: impl Future<Output = Out> + Send + 'static)
where
Out: Send + 'static,
{
cfg_if! {
if #[cfg(feature="rt-async-std")] {
drop(async_std::task::spawn(future));
} else if #[cfg(feature="rt-tokio")] {
drop(tokio::task::spawn(future));
}
}
}
pub fn spawn_detached_local<Out>(future: impl Future<Output = Out> + 'static)
where
Out: 'static,
{
cfg_if! {
if #[cfg(feature="rt-async-std")] {
drop(async_std::task::spawn_local(future));
} else if #[cfg(feature="rt-tokio")] {
drop(tokio::task::spawn_local(future));
}
}
}
pub async fn blocking_wrapper<F, R>(blocking_task: F, err_result: R) -> R
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
// run blocking stuff in blocking thread
cfg_if! {
if #[cfg(feature="rt-async-std")] {
async_std::task::spawn_blocking(blocking_task).await
} else if #[cfg(feature="rt-tokio")] {
tokio::task::spawn_blocking(blocking_task).await.unwrap_or(err_result)
} else {
#[compile_error("must use an executor")]
}
}
}
}
}

391
veilid-tools/src/split_url.rs Normal file
View File

@@ -0,0 +1,391 @@
// Loose subset interpretation of the URL standard
// Not using full Url crate here for no_std compatibility
//
// Caveats:
// No support for query string parsing
// No support for paths with ';' parameters
// URLs must convert to UTF8
// Only IP address and DNS hostname host fields are supported
use super::*;
use core::str::FromStr;
fn is_alphanum(c: u8) -> bool {
matches!(c,
b'A'..=b'Z'
| b'a'..=b'z'
| b'0'..=b'9'
)
}
fn is_mark(c: u8) -> bool {
matches!(
c,
b'-' | b'_' | b'.' | b'!' | b'~' | b'*' | b'\'' | b'(' | b')'
)
}
fn is_unreserved(c: u8) -> bool {
is_alphanum(c) || is_mark(c)
}
fn must_encode_userinfo(c: u8) -> bool {
!(is_unreserved(c) || matches!(c, b'%' | b':' | b';' | b'&' | b'=' | b'+' | b'$' | b','))
}
fn must_encode_path(c: u8) -> bool {
!(is_unreserved(c)
|| matches!(
c,
b'%' | b'/' | b':' | b'@' | b'&' | b'=' | b'+' | b'$' | b','
))
}
fn is_valid_scheme<H: AsRef<str>>(host: H) -> bool {
let mut chars = host.as_ref().chars();
if let Some(ch) = chars.next() {
if !matches!(ch, 'A'..='Z' | 'a'..='z') {
return false;
}
} else {
return false;
}
for ch in chars {
if !matches!(ch,
'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '+' | '.' )
{
return false;
}
}
true
}
fn hex_decode(h: u8) -> Result<u8, SplitUrlError> {
match h {
b'0'..=b'9' => Ok(h - b'0'),
b'A'..=b'F' => Ok(h - b'A' + 10),
b'a'..=b'f' => Ok(h - b'a' + 10),
_ => Err(SplitUrlError::new(
"Unexpected character in percent encoding",
)),
}
}
fn hex_encode(c: u8) -> (char, char) {
let c0 = c >> 4;
let c1 = c & 15;
(
if c0 < 10 {
char::from_u32((b'0' + c0) as u32).unwrap()
} else {
char::from_u32((b'A' + c0 - 10) as u32).unwrap()
},
if c1 < 10 {
char::from_u32((b'0' + c1) as u32).unwrap()
} else {
char::from_u32((b'A' + c1 - 10) as u32).unwrap()
},
)
}
fn url_decode<S: AsRef<str>>(s: S) -> Result<String, SplitUrlError> {
let url = s.as_ref().to_owned();
if !url.is_ascii() {
return Err(SplitUrlError::new("URL is not in ASCII encoding"));
}
let url_bytes = url.as_bytes();
let mut dec_bytes: Vec<u8> = Vec::with_capacity(url_bytes.len());
let mut i = 0;
let end = url_bytes.len();
while i < end {
let mut b = url_bytes[i];
i += 1;
if b == b'%' {
if (i + 1) >= end {
return Err(SplitUrlError::new("Invalid URL encoding"));
}
b = hex_decode(url_bytes[i])? << 4 | hex_decode(url_bytes[i + 1])?;
i += 2;
}
dec_bytes.push(b);
}
String::from_utf8(dec_bytes)
.map_err(|e| SplitUrlError::new(format!("Decoded URL is not valid UTF-8: {}", e)))
}
fn url_encode<S: AsRef<str>>(s: S, must_encode: impl Fn(u8) -> bool) -> String {
let bytes = s.as_ref().as_bytes();
let mut out = String::new();
for b in bytes {
if must_encode(*b) {
let (c0, c1) = hex_encode(*b);
out.push('%');
out.push(c0);
out.push(c1);
} else {
out.push(char::from_u32(*b as u32).unwrap())
}
}
out
}
fn convert_port<N>(port_str: N) -> Result<u16, SplitUrlError>
where
N: AsRef<str>,
{
port_str
.as_ref()
.parse::<u16>()
.map_err(|e| SplitUrlError::new(format!("Invalid port: {}", e)))
}
///////////////////////////////////////////////////////////////////////////////
#[derive(ThisError, Debug, Clone, Eq, PartialEq)]
#[error("SplitUrlError: {0}")]
pub struct SplitUrlError(String);
impl SplitUrlError {
pub fn new<T: ToString>(message: T) -> Self {
SplitUrlError(message.to_string())
}
}
///////////////////////////////////////////////////////////////////////////////
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct SplitUrlPath {
pub path: String,
pub fragment: Option<String>,
pub query: Option<String>,
}
impl SplitUrlPath {
pub fn new<P, F, Q>(path: P, fragment: Option<F>, query: Option<Q>) -> Self
where
P: AsRef<str>,
F: AsRef<str>,
Q: AsRef<str>,
{
Self {
path: path.as_ref().to_owned(),
fragment: fragment.map(|f| f.as_ref().to_owned()),
query: query.map(|f| f.as_ref().to_owned()),
}
}
}
impl FromStr for SplitUrlPath {
type Err = SplitUrlError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
Ok(if let Some((p, q)) = s.split_once('?') {
if let Some((p, f)) = p.split_once('#') {
SplitUrlPath::new(url_decode(p)?, Some(url_decode(f)?), Some(q))
} else {
SplitUrlPath::new(url_decode(p)?, Option::<String>::None, Some(q))
}
} else if let Some((p, f)) = s.split_once('#') {
SplitUrlPath::new(url_decode(p)?, Some(url_decode(f)?), Option::<String>::None)
} else {
SplitUrlPath::new(
url_decode(s)?,
Option::<String>::None,
Option::<String>::None,
)
})
}
}
impl fmt::Display for SplitUrlPath {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if let Some(fragment) = &self.fragment {
if let Some(query) = &self.query {
write!(
f,
"{}#{}?{}",
url_encode(&self.path, must_encode_path),
url_encode(fragment, must_encode_path),
query
)
} else {
write!(f, "{}#{}", self.path, fragment)
}
} else if let Some(query) = &self.query {
write!(f, "{}?{}", url_encode(&self.path, must_encode_path), query)
} else {
write!(f, "{}", url_encode(&self.path, must_encode_path))
}
}
}
///////////////////////////////////////////////////////////////////////////////
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum SplitUrlHost {
Hostname(String),
IpAddr(IpAddr),
}
impl SplitUrlHost {
pub fn new<S: AsRef<str>>(s: S) -> Result<Self, SplitUrlError> {
Self::from_str(s.as_ref())
}
}
impl FromStr for SplitUrlHost {
type Err = SplitUrlError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
if s.is_empty() {
return Err(SplitUrlError::new("Host is empty"));
}
if let Ok(v4) = Ipv4Addr::from_str(s) {
return Ok(SplitUrlHost::IpAddr(IpAddr::V4(v4)));
}
if &s[0..1] == "[" && &s[s.len() - 1..] == "]" {
if let Ok(v6) = Ipv6Addr::from_str(&s[1..s.len() - 1]) {
return Ok(SplitUrlHost::IpAddr(IpAddr::V6(v6)));
}
return Err(SplitUrlError::new("Invalid ipv6 address"));
}
for ch in s.chars() {
if !matches!(ch,
'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '.' )
{
return Err(SplitUrlError::new("Invalid hostname"));
}
}
Ok(SplitUrlHost::Hostname(s.to_owned()))
}
}
impl fmt::Display for SplitUrlHost {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Hostname(h) => {
write!(f, "{}", h)
}
Self::IpAddr(IpAddr::V4(v4)) => {
write!(f, "{}", v4)
}
Self::IpAddr(IpAddr::V6(v6)) => {
write!(f, "[{}]", v6)
}
}
}
}
///////////////////////////////////////////////////////////////////////////////
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct SplitUrl {
pub scheme: String,
pub userinfo: Option<String>,
pub host: SplitUrlHost,
pub port: Option<u16>,
pub path: Option<SplitUrlPath>,
}
impl SplitUrl {
pub fn new<S>(
scheme: S,
userinfo: Option<String>,
host: SplitUrlHost,
port: Option<u16>,
path: Option<SplitUrlPath>,
) -> Self
where
S: AsRef<str>,
{
Self {
scheme: scheme.as_ref().to_owned(),
userinfo,
host,
port,
path,
}
}
pub fn host_port(&self, default_port: u16) -> String {
format!("{}:{}", self.host, self.port.unwrap_or(default_port))
}
}
fn split_host_with_port(s: &str) -> Option<(&str, &str)> {
// special case for ipv6 colons
if s.len() > 2 && s[0..1] == *"[" {
if let Some(end) = s.find(']') {
if end < (s.len() - 2) && s[end + 1..end + 2] == *":" {
return Some((&s[0..end + 1], &s[end + 2..]));
}
}
None
} else {
s.split_once(':')
}
}
impl FromStr for SplitUrl {
type Err = SplitUrlError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
if let Some((scheme, mut rest)) = s.split_once("://") {
if !is_valid_scheme(scheme) {
return Err(SplitUrlError::new("Invalid scheme specified"));
}
let userinfo = {
if let Some((userinfo_str, after)) = rest.split_once('@') {
rest = after;
Some(url_decode(userinfo_str)?)
} else {
None
}
};
if let Some((host, rest)) = split_host_with_port(rest) {
let host = SplitUrlHost::from_str(host)?;
if let Some((portstr, path)) = rest.split_once('/') {
let port = convert_port(portstr)?;
let path = SplitUrlPath::from_str(path)?;
Ok(SplitUrl::new(
scheme,
userinfo,
host,
Some(port),
Some(path),
))
} else {
let port = convert_port(rest)?;
Ok(SplitUrl::new(scheme, userinfo, host, Some(port), None))
}
} else if let Some((host, path)) = rest.split_once('/') {
let host = SplitUrlHost::from_str(host)?;
let path = SplitUrlPath::from_str(path)?;
Ok(SplitUrl::new(scheme, userinfo, host, None, Some(path)))
} else {
let host = SplitUrlHost::from_str(rest)?;
Ok(SplitUrl::new(scheme, userinfo, host, None, None))
}
} else {
Err(SplitUrlError::new("No scheme specified"))
}
}
}
impl fmt::Display for SplitUrl {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let hostname = {
if let Some(userinfo) = &self.userinfo {
let userinfo = url_encode(userinfo, must_encode_userinfo);
if let Some(port) = self.port {
format!("{}@{}:{}", userinfo, self.host, port)
} else {
format!("{}@{}", userinfo, self.host)
}
} else if let Some(port) = self.port {
format!("{}:{}", self.host, port)
} else {
format!("{}", self.host)
}
};
if let Some(path) = &self.path {
write!(f, "{}://{}/{}", self.scheme, hostname, path)
} else {
write!(f, "{}://{}", self.scheme, hostname)
}
}
}

145
veilid-tools/src/tick_task.rs Normal file
View File

@@ -0,0 +1,145 @@
use super::*;
use core::sync::atomic::{AtomicU64, Ordering};
use once_cell::sync::OnceCell;
type TickTaskRoutine<E> =
dyn Fn(StopToken, u64, u64) -> SendPinBoxFuture<Result<(), E>> + Send + Sync + 'static;
/// Runs a single-future background processing task, attempting to run it once every 'tick period' microseconds.
/// If the prior tick is still running, it will allow it to finish, and do another tick when the timer comes around again.
/// One should attempt to make tasks short-lived things that run in less than the tick period if you want things to happen with regular periodicity.
pub struct TickTask<E: Send + 'static> {
last_timestamp_us: AtomicU64,
tick_period_us: u64,
routine: OnceCell<Box<TickTaskRoutine<E>>>,
stop_source: AsyncMutex<Option<StopSource>>,
single_future: MustJoinSingleFuture<Result<(), E>>,
running: Arc<AtomicBool>,
}
impl<E: Send + 'static> TickTask<E> {
pub fn new_us(tick_period_us: u64) -> Self {
Self {
last_timestamp_us: AtomicU64::new(0),
tick_period_us,
routine: OnceCell::new(),
stop_source: AsyncMutex::new(None),
single_future: MustJoinSingleFuture::new(),
running: Arc::new(AtomicBool::new(false)),
}
}
pub fn new_ms(tick_period_ms: u32) -> Self {
Self {
last_timestamp_us: AtomicU64::new(0),
tick_period_us: (tick_period_ms as u64) * 1000u64,
routine: OnceCell::new(),
stop_source: AsyncMutex::new(None),
single_future: MustJoinSingleFuture::new(),
running: Arc::new(AtomicBool::new(false)),
}
}
pub fn new(tick_period_sec: u32) -> Self {
Self {
last_timestamp_us: AtomicU64::new(0),
tick_period_us: (tick_period_sec as u64) * 1000000u64,
routine: OnceCell::new(),
stop_source: AsyncMutex::new(None),
single_future: MustJoinSingleFuture::new(),
running: Arc::new(AtomicBool::new(false)),
}
}
pub fn set_routine(
&self,
routine: impl Fn(StopToken, u64, u64) -> SendPinBoxFuture<Result<(), E>> + Send + Sync + 'static,
) {
self.routine.set(Box::new(routine)).map_err(drop).unwrap();
}
pub fn is_running(&self) -> bool {
self.running.load(core::sync::atomic::Ordering::Relaxed)
}
pub async fn stop(&self) -> Result<(), E> {
// drop the stop source if we have one
let opt_stop_source = &mut *self.stop_source.lock().await;
if opt_stop_source.is_none() {
// already stopped, just return
trace!("tick task already stopped");
return Ok(());
}
drop(opt_stop_source.take());
// wait for completion of the tick task
trace!("stopping single future");
match self.single_future.join().await {
Ok(Some(Err(err))) => Err(err),
_ => Ok(()),
}
}
pub async fn tick(&self) -> Result<(), E> {
let now = get_timestamp();
let last_timestamp_us = self.last_timestamp_us.load(Ordering::Acquire);
if last_timestamp_us != 0u64 && now.saturating_sub(last_timestamp_us) < self.tick_period_us
{
// It's not time yet
return Ok(());
}
// Lock the stop source, tells us if we have ever started this future
let opt_stop_source = &mut *self.stop_source.lock().await;
if opt_stop_source.is_some() {
// See if the previous execution finished with an error
match self.single_future.check().await {
Ok(Some(Err(e))) => {
// We have an error result, which means the singlefuture ran but we need to propagate the error
return Err(e);
}
Ok(Some(Ok(()))) => {
// We have an ok result, which means the singlefuture ran, and we should run it again this tick
}
Ok(None) => {
// No prior result to return which means things are still running
// We can just return now, since the singlefuture will not run a second time
return Ok(());
}
Err(()) => {
// If we get this, it's because we are joining the singlefuture already
// Don't bother running but this is not an error in this case
return Ok(());
}
};
}
// Run the singlefuture
let stop_source = StopSource::new();
let stop_token = stop_source.token();
let running = self.running.clone();
let routine = self.routine.get().unwrap()(stop_token, last_timestamp_us, now);
let wrapped_routine = Box::pin(async move {
running.store(true, core::sync::atomic::Ordering::Relaxed);
let out = routine.await;
running.store(false, core::sync::atomic::Ordering::Relaxed);
out
});
match self.single_future.single_spawn(wrapped_routine).await {
// We should have already consumed the result of the last run, or there was none
// and we should definitely have run, because the prior 'check()' operation
// should have ensured the singlefuture was ready to run
Ok((None, true)) => {
// Set new timer
self.last_timestamp_us.store(now, Ordering::Release);
// Save new stopper
*opt_stop_source = Some(stop_source);
Ok(())
}
// All other conditions should not be reachable
_ => {
unreachable!();
}
}
}
}

32
veilid-tools/src/timeout.rs Normal file
View File

@@ -0,0 +1,32 @@
use super::*;
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
pub async fn timeout<F, T>(dur_ms: u32, f: F) -> Result<T, TimeoutError>
where
F: Future<Output = T>,
{
match select(Box::pin(intf::sleep(dur_ms)), Box::pin(f)).await {
Either::Left((_x, _b)) => Err(TimeoutError()),
Either::Right((y, _a)) => Ok(y),
}
}
} else {
pub async fn timeout<F, T>(dur_ms: u32, f: F) -> Result<T, TimeoutError>
where
F: Future<Output = T>,
{
cfg_if! {
if #[cfg(feature="rt-async-std")] {
async_std::future::timeout(Duration::from_millis(dur_ms as u64), f).await.map_err(|e| e.into())
} else if #[cfg(feature="rt-tokio")] {
tokio::time::timeout(Duration::from_millis(dur_ms as u64), f).await.map_err(|e| e.into())
}
}
}
}
}

177
veilid-tools/src/timeout_or.rs Normal file
View File

@@ -0,0 +1,177 @@
use super::*;
use core::fmt::{Debug, Display};
use core::result::Result;
use std::error::Error;
use std::io;
#[derive(ThisError, Debug, Clone, Copy, Eq, PartialEq)]
#[error("Timeout")]
pub struct TimeoutError();
impl TimeoutError {
pub fn to_io(self) -> io::Error {
io::Error::new(io::ErrorKind::TimedOut, self)
}
}
cfg_if! {
if #[cfg(feature="rt-async-std")] {
impl From<async_std::future::TimeoutError> for TimeoutError {
fn from(_: async_std::future::TimeoutError) -> Self {
Self()
}
}
} else if #[cfg(feature="rt-tokio")] {
impl From<tokio::time::error::Elapsed> for TimeoutError {
fn from(_: tokio::time::error::Elapsed) -> Self {
Self()
}
}
}
}
//////////////////////////////////////////////////////////////////
// Non-fallible timeout conversions
pub trait TimeoutOrExt<T> {
fn into_timeout_or(self) -> TimeoutOr<T>;
}
impl<T> TimeoutOrExt<T> for Result<T, TimeoutError> {
fn into_timeout_or(self) -> TimeoutOr<T> {
self.ok()
.map(|v| TimeoutOr::<T>::Value(v))
.unwrap_or(TimeoutOr::<T>::Timeout)
}
}
pub trait IoTimeoutOrExt<T> {
fn into_timeout_or(self) -> io::Result<TimeoutOr<T>>;
}
impl<T> IoTimeoutOrExt<T> for io::Result<T> {
fn into_timeout_or(self) -> io::Result<TimeoutOr<T>> {
match self {
Ok(v) => Ok(TimeoutOr::<T>::Value(v)),
Err(e) if e.kind() == io::ErrorKind::TimedOut => Ok(TimeoutOr::<T>::Timeout),
Err(e) => Err(e),
}
}
}
pub trait TimeoutOrResultExt<T, E> {
fn into_result(self) -> Result<TimeoutOr<T>, E>;
}
impl<T, E> TimeoutOrResultExt<T, E> for TimeoutOr<Result<T, E>> {
fn into_result(self) -> Result<TimeoutOr<T>, E> {
match self {
TimeoutOr::<Result<T, E>>::Timeout => Ok(TimeoutOr::<T>::Timeout),
TimeoutOr::<Result<T, E>>::Value(Ok(v)) => Ok(TimeoutOr::<T>::Value(v)),
TimeoutOr::<Result<T, E>>::Value(Err(e)) => Err(e),
}
}
}
//////////////////////////////////////////////////////////////////
// Non-fallible timeout
#[must_use]
pub enum TimeoutOr<T> {
Timeout,
Value(T),
}
impl<T> TimeoutOr<T> {
pub fn timeout() -> Self {
Self::Timeout
}
pub fn value(value: T) -> Self {
Self::Value(value)
}
pub fn is_timeout(&self) -> bool {
matches!(self, Self::Timeout)
}
pub fn is_value(&self) -> bool {
matches!(self, Self::Value(_))
}
pub fn map<X, F: Fn(T) -> X>(self, f: F) -> TimeoutOr<X> {
match self {
Self::Timeout => TimeoutOr::<X>::Timeout,
Self::Value(v) => TimeoutOr::<X>::Value(f(v)),
}
}
pub fn on_timeout<F: Fn()>(self, f: F) -> Self {
match self {
Self::Timeout => {
f();
Self::Timeout
}
Self::Value(v) => Self::Value(v),
}
}
pub fn into_timeout_error(self) -> Result<T, TimeoutError> {
match self {
Self::Timeout => Err(TimeoutError {}),
Self::Value(v) => Ok(v),
}
}
pub fn into_option(self) -> Option<T> {
match self {
Self::Timeout => None,
Self::Value(v) => Some(v),
}
}
}
impl<T> From<TimeoutOr<T>> for Option<T> {
fn from(t: TimeoutOr<T>) -> Self {
match t {
TimeoutOr::<T>::Timeout => None,
TimeoutOr::<T>::Value(v) => Some(v),
}
}
}
impl<T: Clone> Clone for TimeoutOr<T> {
fn clone(&self) -> Self {
match self {
Self::Timeout => Self::Timeout,
Self::Value(t) => Self::Value(t.clone()),
}
}
}
impl<T: Debug> Debug for TimeoutOr<T> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Timeout => write!(f, "Timeout"),
Self::Value(arg0) => f.debug_tuple("Value").field(arg0).finish(),
}
}
}
impl<T: Display> Display for TimeoutOr<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Timeout => write!(f, ""),
Self::Value(arg0) => write!(f, "{}", arg0),
}
}
}
impl<T: Debug + Display> Error for TimeoutOr<T> {}
//////////////////////////////////////////////////////////////////
// Non-fallible timeoue macros
#[macro_export]
macro_rules! timeout_or_try {
($r: expr) => {
match $r {
TimeoutOr::Timeout => return Ok(TimeoutOr::Timeout),
TimeoutOr::Value(v) => v,
}
};
}

25
veilid-tools/src/timestamp.rs Normal file
View File

@@ -0,0 +1,25 @@
use super::*;
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
use js_sys::Date;
pub fn get_timestamp() -> u64 {
if utils::is_browser() {
return (Date::now() * 1000.0f64) as u64;
} else {
panic!("WASM requires browser environment");
}
}
} else {
use std::time::{SystemTime, UNIX_EPOCH};
pub fn get_timestamp() -> u64 {
match SystemTime::now().duration_since(UNIX_EPOCH) {
Ok(n) => n.as_micros() as u64,
Err(_) => panic!("SystemTime before UNIX_EPOCH!"),
}
}
}
}

301
veilid-tools/src/tools.rs Normal file
View File

@@ -0,0 +1,301 @@
use super::*;
use alloc::string::ToString;
use std::io;
use std::path::Path;
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
#[macro_export]
macro_rules! assert_err {
($ex:expr) => {
if let Ok(v) = $ex {
panic!("assertion failed, expected Err(..), got {:?}", v);
}
};
}
#[macro_export]
macro_rules! io_error_other {
($msg:expr) => {
io::Error::new(io::ErrorKind::Other, $msg.to_string())
};
}
pub fn to_io_error_other<E: std::error::Error + Send + Sync + 'static>(x: E) -> io::Error {
io::Error::new(io::ErrorKind::Other, x)
}
#[macro_export]
macro_rules! bail_io_error_other {
($msg:expr) => {
return io::Result::Err(io::Error::new(io::ErrorKind::Other, $msg.to_string()))
};
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
pub fn system_boxed<'a, Out>(
future: impl Future<Output = Out> + Send + 'a,
) -> SendPinBoxFutureLifetime<'a, Out> {
Box::pin(future)
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
// xxx: for now until wasm threads are more stable, and/or we bother with web workers
pub fn get_concurrency() -> u32 {
1
}
} else {
pub fn get_concurrency() -> u32 {
std::thread::available_parallelism()
.map(|x| x.get())
.unwrap_or_else(|e| {
warn!("unable to get concurrency defaulting to single core: {}", e);
1
}) as u32
}
}
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
pub fn split_port(name: &str) -> EyreResult<(String, Option<u16>)> {
if let Some(split) = name.rfind(':') {
let hoststr = &name[0..split];
let portstr = &name[split + 1..];
let port: u16 = portstr.parse::<u16>().wrap_err("invalid port")?;
Ok((hoststr.to_string(), Some(port)))
} else {
Ok((name.to_string(), None))
}
}
pub fn prepend_slash(s: String) -> String {
if s.starts_with('/') {
return s;
}
let mut out = "/".to_owned();
out.push_str(s.as_str());
out
}
pub fn timestamp_to_secs(ts: u64) -> f64 {
ts as f64 / 1000000.0f64
}
pub fn secs_to_timestamp(secs: f64) -> u64 {
(secs * 1000000.0f64) as u64
}
pub fn ms_to_us(ms: u32) -> u64 {
(ms as u64) * 1000u64
}
// Calculate retry attempt with logarhythmic falloff
pub fn retry_falloff_log(
last_us: u64,
cur_us: u64,
interval_start_us: u64,
interval_max_us: u64,
interval_multiplier_us: f64,
) -> bool {
//
if cur_us < interval_start_us {
// Don't require a retry within the first 'interval_start_us' microseconds of the reliable time period
false
} else if cur_us >= last_us + interval_max_us {
// Retry at least every 'interval_max_us' microseconds
true
} else {
// Exponential falloff between 'interval_start_us' and 'interval_max_us' microseconds
last_us <= secs_to_timestamp(timestamp_to_secs(cur_us) / interval_multiplier_us)
}
}
pub fn try_at_most_n_things<T, I, C, R>(max: usize, things: I, closure: C) -> Option<R>
where
I: IntoIterator<Item = T>,
C: Fn(T) -> Option<R>,
{
let mut fails = 0usize;
for thing in things.into_iter() {
if let Some(r) = closure(thing) {
return Some(r);
}
fails += 1;
if fails >= max {
break;
}
}
None
}
pub async fn async_try_at_most_n_things<T, I, C, R, F>(
max: usize,
things: I,
closure: C,
) -> Option<R>
where
I: IntoIterator<Item = T>,
C: Fn(T) -> F,
F: Future<Output = Option<R>>,
{
let mut fails = 0usize;
for thing in things.into_iter() {
if let Some(r) = closure(thing).await {
return Some(r);
}
fails += 1;
if fails >= max {
break;
}
}
None
}
pub trait CmpAssign {
fn min_assign(&mut self, other: Self);
fn max_assign(&mut self, other: Self);
}
impl<T> CmpAssign for T
where
T: core::cmp::Ord,
{
fn min_assign(&mut self, other: Self) {
if &other < self {
*self = other;
}
}
fn max_assign(&mut self, other: Self) {
if &other > self {
*self = other;
}
}
}
pub fn compatible_unspecified_socket_addr(socket_addr: &SocketAddr) -> SocketAddr {
match socket_addr {
SocketAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0),
SocketAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0)), 0),
}
}
pub fn listen_address_to_socket_addrs(listen_address: &str) -> EyreResult<Vec<SocketAddr>> {
// If no address is specified, but the port is, use ipv4 and ipv6 unspecified
// If the address is specified, only use the specified port and fail otherwise
let ip_addrs = vec![
IpAddr::V4(Ipv4Addr::UNSPECIFIED),
IpAddr::V6(Ipv6Addr::UNSPECIFIED),
];
Ok(if let Some(portstr) = listen_address.strip_prefix(':') {
let port = portstr
.parse::<u16>()
.wrap_err("Invalid port format in udp listen address")?;
ip_addrs.iter().map(|a| SocketAddr::new(*a, port)).collect()
} else if let Ok(port) = listen_address.parse::<u16>() {
ip_addrs.iter().map(|a| SocketAddr::new(*a, port)).collect()
} else {
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
use core::str::FromStr;
vec![SocketAddr::from_str(listen_address).map_err(|e| io_error_other!(e)).wrap_err("Unable to parse address")?]
} else {
listen_address
.to_socket_addrs()
.wrap_err("Unable to resolve address")?
.collect()
}
}
})
}
pub trait Dedup<T: PartialEq + Clone> {
fn remove_duplicates(&mut self);
}
impl<T: PartialEq + Clone> Dedup<T> for Vec<T> {
fn remove_duplicates(&mut self) {
let mut already_seen = Vec::new();
self.retain(|item| match already_seen.contains(item) {
true => false,
_ => {
already_seen.push(item.clone());
true
}
})
}
}
cfg_if::cfg_if! {
if #[cfg(unix)] {
use std::os::unix::fs::MetadataExt;
use std::os::unix::prelude::PermissionsExt;
use nix::unistd::{Uid, Gid};
pub fn ensure_file_private_owner<P:AsRef<Path>>(path: P) -> EyreResult<()>
{
let path = path.as_ref();
if !path.exists() {
return Ok(());
}
let uid = Uid::effective();
let gid = Gid::effective();
let meta = std::fs::metadata(path).wrap_err("unable to get metadata for path")?;
if meta.mode() != 0o600 {
std::fs::set_permissions(path,std::fs::Permissions::from_mode(0o600)).wrap_err("unable to set correct permissions on path")?;
}
if meta.uid() != uid.as_raw() || meta.gid() != gid.as_raw() {
bail!("path has incorrect owner/group");
}
Ok(())
}
} else if #[cfg(windows)] {
//use std::os::windows::fs::MetadataExt;
//use windows_permissions::*;
pub fn ensure_file_private_owner<P:AsRef<Path>>(path: P) -> EyreResult<()>
{
let path = path.as_ref();
if !path.exists() {
return Ok(());
}
Ok(())
}
} else {
pub fn ensure_file_private_owner<P:AsRef<Path>>(_path: P) -> Result<(),String>
{
Ok(())
}
}
}
#[repr(C, align(8))]
struct AlignToEight([u8; 8]);
pub unsafe fn aligned_8_u8_vec_uninit(n_bytes: usize) -> Vec<u8> {
let n_units = (n_bytes + mem::size_of::<AlignToEight>() - 1) / mem::size_of::<AlignToEight>();
let mut aligned: Vec<AlignToEight> = Vec::with_capacity(n_units);
let ptr = aligned.as_mut_ptr();
let cap_units = aligned.capacity();
mem::forget(aligned);
Vec::from_raw_parts(
ptr as *mut u8,
n_bytes,
cap_units * mem::size_of::<AlignToEight>(),
)
}

52
veilid-tools/src/wasm.rs Normal file
View File

@@ -0,0 +1,52 @@
use super::*;
use core::sync::atomic::{AtomicI8, Ordering};
use js_sys::{global, Reflect};
#[wasm_bindgen]
extern "C" {
// Use `js_namespace` here to bind `console.log(..)` instead of just
// `log(..)`
#[wasm_bindgen(js_namespace = console, js_name = log)]
pub fn console_log(s: &str);
#[wasm_bindgen]
pub fn alert(s: &str);
}
pub fn is_browser() -> bool {
static CACHE: AtomicI8 = AtomicI8::new(-1);
let cache = CACHE.load(Ordering::Relaxed);
if cache != -1 {
return cache != 0;
}
let res = Reflect::has(&global().as_ref(), &"window".into()).unwrap_or_default();
CACHE.store(res as i8, Ordering::Relaxed);
res
}
// pub fn is_browser_https() -> bool {
// static CACHE: AtomicI8 = AtomicI8::new(-1);
// let cache = CACHE.load(Ordering::Relaxed);
// if cache != -1 {
// return cache != 0;
// }
// let res = js_sys::eval("window.location.protocol === 'https'")
// .map(|res| res.is_truthy())
// .unwrap_or_default();
// CACHE.store(res as i8, Ordering::Relaxed);
// res
// }
#[derive(ThisError, Debug, Clone, Eq, PartialEq)]
#[error("JsValue error")]
pub struct JsValueError(String);
pub fn map_jsvalue_error(x: JsValue) -> JsValueError {
JsValueError(x.as_string().unwrap_or_default())
}