add gat & tait

Author: smallnest

Cargo.toml

@@ -1,2 +1,2 @@
 [workspace]
-members = ["thread", "pool", "asyncwait", "primitive", "collections", "process", "syncs", "channel", "timer_examples", "parking_lot_examples", "crossbeam_examples", "rayon_examples", "tokio_examples", "special"]
+members = ["thread", "pool", "asyncwait", "container_primitive", "sync_primitive", "collections", "process",  "channel", "timer_examples", "parking_lot_examples", "crossbeam_examples", "rayon_examples", "tokio_examples", "special"]

asyncwait/Cargo.toml

@@ -9,6 +9,8 @@ edition = "2021"
 anyhow = "1.0.65"
 async-std = "1.12.0"
 async-stream = "0.3.3"
+async-trait = "0.1.57"
+bytes = "1.2.1"
 futures = { version = "0.3.24", features = ["executor", "thread-pool"] } 
 futures-lite = "1.12.0"
 futures-util = "0.3.24"

asyncwait/src/async_trait_example.rs

@@ -0,0 +1,24 @@
+use async_trait::*; // the official feature is not stable yet
+
+// https://blog.theincredibleholk.org/blog/2022/04/18/how-async-functions-in-traits-could-work-in-rustc/
+
+#[async_trait]
+trait AsyncTrait {
+    async fn get_string(&self) -> String;
+}
+
+#[async_trait]
+impl AsyncTrait for i32 {
+    async fn get_string(&self) -> String {
+        self.to_string()
+    }
+}
+
+pub fn async_trait_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.block_on(async {
+        let x = 10;
+        let y = x.get_string().await;
+        println!("y={}", y);
+    });
+}

asyncwait/src/gat.rs

@@ -0,0 +1,76 @@
+use std::future::Future;
+use async_std::io::WriteExt;
+use bytes::Bytes;
+
+// copy from https://www.sobyte.net/post/2022-04/rust-gat-async-trait/
+//
+// generic_associated_types
+// type_alias_impl_trait traits
+pub trait KvIterator {
+    type NextFuture<'a>: Future<Output = Option<(&'a [u8], &'a [u8])>>
+    where
+        Self: 'a;
+
+    /// Get the next item from the iterator.
+    fn next(&mut self) -> Self::NextFuture<'_>;
+}
+
+pub struct TestIterator {
+    idx: usize,
+    to_idx: usize,
+    key: Vec<u8>,
+    value: Vec<u8>,
+}
+
+impl TestIterator {
+    pub fn new(from_idx: usize, to_idx: usize) -> Self {
+        Self {
+            idx: from_idx,
+            to_idx,
+            key: Vec::new(),
+            value: Vec::new(),
+        }
+    }
+}
+
+#[allow(deprecated_where_clause_location)]
+impl KvIterator for TestIterator {
+    type NextFuture<'a>
+    where
+        Self: 'a,
+    = impl Future<Output = Option<(&'a [u8], &'a [u8])>>;
+
+    fn next(&mut self) -> Self::NextFuture<'_> {
+        async move {
+            if self.idx >= self.to_idx {
+                return None;
+            }
+
+            // Zero-allocation key value manipulation
+
+            self.key.clear();
+
+            write!(&mut self.key, "key_{:05}", self.idx).await.unwrap();
+
+            self.value.clear();
+            write!(&mut self.value, "value_{:05}", self.idx).await.unwrap();
+
+            self.idx += 1;
+            Some((&self.key[..], &self.value[..]))
+        }
+    }
+}
+
+pub fn kviterator_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.block_on(async {
+        let mut iter = TestIterator::new(0, 10);
+        while let Some((key, value)) = iter.next().await {
+            println!(
+                "{:?} {:?}",
+                Bytes::copy_from_slice(key),
+                Bytes::copy_from_slice(value)
+            );
+        }
+    });
+}

asyncwait/src/lib.rs

@@ -1,100 +1,15 @@
+#![feature(type_alias_impl_trait)]
+// #![feature(return_position_impl_trait_in_trait)]
+
 pub mod asyncio;
 pub mod future;
+pub mod runtimes;
+pub mod gat;
+pub mod async_trait_example;
 
 pub use asyncio::*;
 pub use future::*;
+pub use runtimes::*;
+pub use gat::*;
+pub use async_trait_example::*; 
 
-use futures::channel::mpsc;
-use futures::executor::{self, ThreadPool};
-use futures::try_join;
-use futures::StreamExt;
-use futures::{
-    future::FutureExt, // for `.fuse()`
-    pin_mut,
-    select,
-};
-
-pub fn tokio_async() {
-    let rt = tokio::runtime::Runtime::new().unwrap();
-    rt.block_on(async {
-        println!("Hello from tokio!");
-
-        rt.spawn(async {
-            println!("Hello from a tokio task!");
-            println!("in spawn")
-        })
-        .await
-        .unwrap();
-    });
-
-    rt.spawn_blocking(|| println!("in spawn_blocking"));
-}
-
-pub fn futures_async() {
-    let pool = ThreadPool::new().expect("Failed to build pool");
-    let (tx, rx) = mpsc::unbounded::<i32>();
-
-    let fut_values = async {
-        let fut_tx_result = async move {
-            (0..100).for_each(|v| {
-                tx.unbounded_send(v).expect("Failed to send");
-            })
-        };
-        pool.spawn_ok(fut_tx_result);
-
-        let fut_values = rx.map(|v| v * 2).collect();
-
-        fut_values.await
-    };
-
-    let values: Vec<i32> = executor::block_on(fut_values);
-
-    println!("Values={:?}", values);
-}
-
-pub fn futures_lite_async() {
-    futures_lite::future::block_on(async { println!("Hello from futures_lite") })
-}
-
-pub fn async_std() {
-    async_std::task::block_on(async { println!("Hello from async_std") })
-}
-
-pub fn smol_async() {
-    smol::block_on(async { println!("Hello from smol") })
-}
-
-struct Book();
-struct Music();
-
-async fn get_book() -> Result<Book, String> {
-    println!("in get_book");
-    Ok(Book())
-}
-async fn get_music() -> Result<Music, String> {
-    println!("in get_music");
-    Ok(Music())
-}
-async fn get_book_and_music() -> Result<(Book, Music), String> {
-    let book_fut = get_book();
-    let music_fut = get_music();
-    try_join!(book_fut, music_fut)
-}
-
-pub fn join() {
-    futures_lite::future::block_on(async { get_book_and_music().await }).unwrap();
-}
-
-pub fn select() {
-    futures_lite::future::block_on(async {
-        let t1 = get_book().fuse();
-        let t2 = get_music().fuse();
-
-        pin_mut!(t1, t2);
-
-        select! {
-            _x = t1 => println!("select get_book"),
-            _y = t2 => println!("select get_music"),
-        }
-    });
-}

asyncwait/src/main.rs

@@ -14,5 +14,7 @@ fn main() {
 
     stream();
 
+    kviterator_example();
 
+    async_trait_example();
 }

asyncwait/src/runtimes.rs

@@ -0,0 +1,96 @@
+
+use futures::channel::mpsc;
+use futures::executor::{self, ThreadPool};
+use futures::try_join;
+use futures::StreamExt;
+use futures::{
+    future::FutureExt, // for `.fuse()`
+    pin_mut,
+    select,
+};
+
+
+pub fn tokio_async() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.block_on(async {
+        println!("Hello from tokio!");
+
+        rt.spawn(async {
+            println!("Hello from a tokio task!");
+            println!("in spawn")
+        })
+        .await
+        .unwrap();
+    });
+
+    rt.spawn_blocking(|| println!("in spawn_blocking"));
+}
+
+pub fn futures_async() {
+    let pool = ThreadPool::new().expect("Failed to build pool");
+    let (tx, rx) = mpsc::unbounded::<i32>();
+
+    let fut_values = async {
+        let fut_tx_result = async move {
+            (0..100).for_each(|v| {
+                tx.unbounded_send(v).expect("Failed to send");
+            })
+        };
+        pool.spawn_ok(fut_tx_result);
+
+        let fut_values = rx.map(|v| v * 2).collect();
+
+        fut_values.await
+    };
+
+    let values: Vec<i32> = executor::block_on(fut_values);
+
+    println!("Values={:?}", values);
+}
+
+pub fn futures_lite_async() {
+    futures_lite::future::block_on(async { println!("Hello from futures_lite") })
+}
+
+pub fn async_std() {
+    async_std::task::block_on(async { println!("Hello from async_std") })
+}
+
+pub fn smol_async() {
+    smol::block_on(async { println!("Hello from smol") })
+}
+
+struct Book();
+struct Music();
+
+async fn get_book() -> Result<Book, String> {
+    println!("in get_book");
+    Ok(Book())
+}
+async fn get_music() -> Result<Music, String> {
+    println!("in get_music");
+    Ok(Music())
+}
+async fn get_book_and_music() -> Result<(Book, Music), String> {
+    let book_fut = get_book();
+    let music_fut = get_music();
+    try_join!(book_fut, music_fut)
+}
+
+pub fn join() {
+    futures_lite::future::block_on(async { get_book_and_music().await }).unwrap();
+}
+
+pub fn select() {
+    futures_lite::future::block_on(async {
+        let t1 = get_book().fuse();
+        let t2 = get_music().fuse();
+
+        pin_mut!(t1, t2);
+
+        select! {
+            _x = t1 => println!("select get_book"),
+            _y = t2 => println!("select get_music"),
+        }
+    });
+}

channel/Cargo.toml

@@ -6,3 +6,11 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+async-channel = "1.7.1"
+async-priority-channel = "0.1.0"
+atomic_mpmc = "0.2.0"
+broadcaster = "1.0.0"
+crossfire = "0.1.7"
+flume = "0.10.14"
+futures-util = "0.3.24"
+tokio = { version = "1.21.2", features = ["full"] }