use core::mem::MaybeUninit;
use core::slice;
use core::time::Duration;
use hirun::channel::{mpsc, spmc, spsc};
use hirun::net::*;
use hirun::runtime::{self, *};
use hirun::thread;
use std::time::Instant;
fn main() {
Builder::new().build().unwrap();
//unsafe { Builder::new(0).rebuild().unwrap() };
let n = 260;
test_socket(2000, n, 1000, 2000);
//return;
test_sync(n, 1000);
//return;
test_channel();
test_futures();
test_mutex();
test_attr();
}
fn to_uninit<T>(datas: &[T]) -> &[MaybeUninit<T>] {
unsafe {
let ptr = &datas[0] as *const T as *const MaybeUninit<T>;
slice::from_raw_parts(ptr, datas.len())
}
}
fn test_channel() {
let cnt = 100000;
println!("+++ Run spsc");
let (sender, receiver) = spsc::channel::<i32>(10).unwrap();
let handle = std::thread::spawn(move || {
for i in 0..cnt {
sender.send(i);
}
});
for _ in 0..cnt {
let data = receiver.recv();
assert!(data >= 0 || data < cnt);
}
assert_eq!(None, receiver.try_recv());
handle.join().unwrap();
println!("--- End");
println!("+++ Run mpsc");
let (sender1, receiver) = mpsc::channel::<i32>(5).unwrap();
let sender2 = sender1.clone();
let handle1 = thread::spawn(move || {
let datas = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
thread::set_cpu(0);
println!("cpu count {}", thread::get_cpu_count());
sender1.send_slice(to_uninit(&datas[..]));
});
let handle2 = thread::spawn(move || {
let datas = [11, 12, 13, 14, 15, 16, 17, 18, 19, 20];
thread::set_cpu(1);
println!("cpu count {}", thread::get_cpu_count());
sender2.send_slice(to_uninit(&datas[..]));
});
for _ in 0..21 {
let data = receiver.recv();
assert!((0..21).contains(&data));
}
println!("cpu count {}", thread::get_cpu_count());
assert_eq!(None, receiver.try_recv());
handle1.join().unwrap();
handle2.join().unwrap();
println!("--- End");
println!("+++ Run spmc");
let (sender, receiver1) = spmc::channel::<i32>(10).unwrap();
let receiver2 = receiver1.clone();
let handle1 = std::thread::spawn(move || {
let mut old = -1;
while let Some(val) = receiver1.recv_timeout(Duration::new(1, 0)) {
assert!(val >= 0 && val < cnt);
old = val;
}
println!("thread1 exit old = {old}");
});
let handle2 = std::thread::spawn(move || {
let mut old = -1;
while let Some(val) = receiver2.recv_timeout(Duration::new(1, 0)) {
assert!(val >= 0 && val < cnt);
old = val;
}
println!("thread2 exit old = {old}");
});
for i in 0..cnt {
sender.send(i);
}
handle1.join().unwrap();
handle2.join().unwrap();
println!("--- End");
}
fn test_futures() {
async fn other_foo() -> i32 {
println!("other_foo");
100
}
async fn sleep_foo(val: i32) -> i32 {
println!("before sleep {:?}", Instant::now());
sleep(Duration::new(10, 0)).await;
println!("after sleep {:?}", Instant::now());
let handle = spawn_with(other_foo(), Attr::default().priority(1));
yield_now().await;
println!("after yield");
let val = handle.await.unwrap() + val;
println!("handle.await");
val
}
let val = spawn(sleep_foo(100)).join().unwrap();
println!("sleep_foo return {val}");
}
fn test_mutex() {
unsafe { Builder::new().rebuild().unwrap() };
let mutex = Mutex::new(1);
let mutex = unsafe { &*(&mutex as *const Mutex<i32>) };
async fn foo(mutex: &Mutex<i32>) -> i32 {
for _ in 0..2 {
let mut guard = mutex.lock().await;
println!("guard: {} {}", *guard, unsafe { libc::pthread_self() });
*guard += 1
}
let guard = mutex.lock().await;
*guard
}
let h1 = spawn(foo(mutex));
let h2 = spawn(foo(mutex));
let h3 = spawn(foo(mutex));
let val = h1.join().unwrap() + h2.join().unwrap() + h3.join().unwrap();
println!("val: {val}, mutex: {}", *mutex.lock_blocking().unwrap());
}
fn test_attr() {
async fn delay() -> i32 {
let tm = Instant::now();
println!("foo now: {tm:?}");
100
}
let h = spawn_with(delay(), Attr::default().delay(Duration::new(5, 0)));
let tm = Instant::now();
println!("spawn delay now: {tm:?}");
let _ = h.join();
async fn timeout() -> i32 {
let tm = Instant::now();
println!("timeout: before sleep(5) , now: {tm:?}");
sleep(Duration::new(5, 0)).await;
let tm = Instant::now();
println!("timeout: after sleep now: {tm:?}");
100
}
let h = spawn_with(
timeout(),
Attr::default().deadline(Duration::new(4, 0)),
);
let tm = Instant::now();
println!("++++ spawn timeout now: {tm:?}");
let ret = h.join();
println!("timeout.join {ret:?}");
println!("--- time::now() {:?}", hirun::time::now());
async fn bar() -> i32 {
100
}
println!("sizeof(bar) = {}", core::mem::size_of_val(&bar()));
async fn baz() {}
println!("sizeof(baz) = {}", core::mem::size_of_val(&baz()));
}
async fn value() -> i32 {
100
}
async fn more_values(cnt: usize) -> i32 {
let mut handles = vec![];
for _ in 0..cnt {
handles.push(runtime::spawn(value()));
}
let mut val = 0;
for handle in handles {
val += handle.await.unwrap();
}
val
}
fn test_sync(cnt: usize, n2: usize) {
//let mut handles = vec![];
//for _ in 0..cnt {
let mut set = runtime::JoinSet::new();
for n in 0..cnt {
let _ = set.spawn(more_values(n + n2));
//handles.push(runtime::spawn(more_values(n2)));
}
runtime::block_on(async move {
for (id, ret) in set.wait_all().await {
println!("{id}: {ret:?}");
}
for n in 0..200 {
let _ = set.spawn(more_values(n + 1));
//handles.push(runtime::spawn(more_values(n2)));
}
while let Some((id, ret)) = set.wait_any().await {
println!("any {id}: {ret:?}");
}
//for handle in handles {
// let _ = handle.await;
//}
})
.unwrap();
}
async fn read(conn: &mut AioFd<'_>, cnt: usize) -> usize {
let mut buf = [0_u8; 100];
let mut recved = 0;
while recved < cnt {
match conn.read(&mut buf).await {
Ok(n) => {
if n == 0 {
break;
}
recved += n;
}
Err(_) => break,
}
}
if recved != cnt {
println!("recv {cnt} bytes return {recved}");
}
recved
}
async fn write(conn: &mut AioFd<'_>, cnt: usize) -> usize {
let buf = [0_u8; 100];
let mut sended = 0;
while sended < cnt {
let len = core::cmp::min(cnt - sended, buf.len());
match conn.write(&buf[..len]).await {
Ok(n) => sended += n,
Err(_) => break,
}
}
if sended != cnt {
println!("write {cnt} bytes return {sended}");
}
sended
}
async fn server_conn(conn: Fd, req: usize, rsp: usize) {
let mut aio = AioFd::new(&conn);
if read(&mut aio, req).await != req {
return;
}
if write(&mut aio, rsp).await != rsp {
//return;
}
}
async fn server(port: u16, mut cnt: usize, req: usize, rsp: usize) {
let addr = SocketAddr::inet("127.0.0.1", port).unwrap();
println!("start server run...{cnt}");
let server = match Fd::tcp_server(&addr) {
Ok(server) => server,
Err(e) => {
println!("start server error: {e}!");
return;
}
};
let mut aio = AioFd::new(&server);
println!("server run...{cnt}");
while cnt > 0 {
println!("server cnt {cnt}");
let Ok((conn, _)) = aio.accept().await else { return; };
cnt -= 1;
runtime::spawn(server_conn(conn, req, rsp));
println!("server left conns: {cnt}");
}
}
async fn client(port: u16, req: usize, rsp: usize) {
let addr = SocketAddr::inet("127.0.0.1", port).unwrap();
let Ok(client) = Fd::tcp_client(addr.family(), None) else { return; };
let mut aio = AioFd::new(&client);
println!("client run...");
if aio.connect(&addr).await.is_err() {
return;
}
if write(&mut aio, req).await != req {
return;
}
if read(&mut aio, rsp).await != rsp {
//return;
}
}
fn test_socket(port: u16, cnt: usize, req_size: usize, resp_size: usize) {
println!("test_socket conn = {cnt} port = {port}");
let server = runtime::spawn(server(port, cnt, req_size, resp_size));
let _ = runtime::block_on(async {
let mut cnt = cnt;
println!("block_on {cnt}");
while cnt > 0 {
let conns = core::cmp::min(cnt, 500);
println!("block_on {cnt} {conns}");
cnt -= conns;
let mut handles = vec![];
for _ in 0..conns {
handles.push(runtime::spawn(client(port, req_size, resp_size)));
}
println!("handles.join()");
for h in handles {
let _ = h.await;
}
println!("conns closed left {cnt}");
}
});
println!("wait server exit");
let _ = server.join();
}