1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

                                 +-+ 阻 塞 I/O (BIO)
                                 |
                                 +-+ 非 阻 塞 I/O (NIO)
                                 |
              +----+ 同 步 I/O +--+
              |                  |
              |                  +-+ I/O 多 路 复 用
              |                  |
              |                  +-+ 信 号 驱 动 I/O
I/O 模 型  +---+
              |
              |
              |                  +-+ Linux (AIO)
              |                  |         (io_uring)
              +----+ 异 步 I/O +--+
                                 |
                                 +-+ windows (IOCP)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

Application               kernel
+---------+            +-----------+  +---+
|         |   syscall  | no        |      |
|   Read  | +--------> | datagram  |      |
| recvfrom|            | ready     |      |
|         |            |    +      |      +-+ wait for
|         |            |    |      |      +-+ data
|         |            |    v      |      |
|         |            | datagram  |      |
|         |            | ready     |  +---+
|         |            |           |
|         |            | copy      |  +---+
|         |            | datagram  |      |
|process  |            |    +      |      +-+ copy data
|datagram |   return   |    |      |      +-+ from kernel to user
|         | <--------+ |    v      |      |
|         |            |  copy     |  +---+
|         |            |  complete |
+---------+            +-----------+

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

                        +--------------------------------+     +-------------------------+
                        | epoll_ctl                      |     | epoll_wait              |
                        |                                |     |                         |
                        |                                |     |         +----+          |
                        |                 +---+          |     |         |    |          |
                        |                 |   |          |     |         |    |          |
                        |               +-+---+--+       |     |         +--+-+          |
                        |               |        |       |     |            |            |
                        |            +--++     +-++      |     |            |            |
epoll_create  +---->    |            |   |     |  |      |     |         +--+-+          |
                        |            +-+-+     +--+      +---->+         |    |          |
                        |              |                 |event|         |    |          |
                        |         +----+--+              |     |         +--+-+          |
                        |         |       |              |     |            |            |
                        |         ++      |              |     |            |            |
                        |        +--+   +-+-+            |     |         +--+-+          |
                        |        |  |   |   |            |     |         |    |          |
                        |        +--+   +---+            |     |         |    |          |
                        |                                |     |         +----+          |
                        |                    红 黑 树     |     |                 链 表    |
                        +--------------------------------+     +-------------------------+

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

+----+ Head  +---------+               +----------+ Head
|            |         |               |          |
|            |         |               |          |
|            +---------+               +----------+
|            |         |               |          |
|            |         |               |          |
|            +---------+               +----------+
|            |         |               |          |
|            |         |               |          |
|            +---------+               +----------+
|            |         |               |          |
|      Tail  +---------+               +----------+ Tail <--+
|        +--------------------------------------------+     |
|        | Kernel                                     |     |
|        |                                            |     |
|        |        +-------+              +-------+    |     |
|        |        |       |              |       |    |     |
+---------------> | SQ    |              |  CQ   | +--------+
         |        |       |              |       |    |
         |        +-------+              +-------+    |
         |                                            |
         +--------------------------------------------+

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

                                                     +----------------+
req                                        Dispatch  |                |
+------+                                  +--------> | req handler    |
|      |                                  |          +----------------+
|      | +----+                           |
+------+      | event    +------------+   |
              |          |            |   |
              +--------> |  Service   |   |Dispatch  +----------------+
                         |  Handler   +------------> |                |
req          +---------> |            |   |          | req handler    |
+------+     |           +------------+   |          +----------------+
|      |     | event                      |
|      +----+                             |
+------+                                  | Dispatch +----------------+
                                          +--------->+                |
                                                     | req handler    |
                                                     +----------------+

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

#[cfg(target_os = "windows")]
mod windows;
#[cfg(target_os = "windows")]
pub use windows::{Event, Registrator, Selector, TcpStream};

#[cfg(target_os = "macos")]
mod macos;
#[cfg(target_os = "macos")]
pub use macos::{Event, Registrator, Selector, TcpStream};

#[cfg(target_os = "linux")]
mod linux;
#[cfg(target_os = "linux")]
pub use linux::{Event, Registrator, Selector, TcpStream};

1
2
3
4
5

#[derive(Debug)]
pub struct Poll {
    registry: Registry,
    is_poll_dead: Arc<AtomicBool>,
}

1
2
3

pub fn poll(&mut self, events: &mut Events, timeout_ms: Option<i32>) -> io::Result<usize>{
  // ...
}

1
2
3

pub struct Poll {
    registry: Registry,
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

A
+------+
|      |
|      |
|      +-------------------+
|      |                   |
|      |                   |
|      |                  Bv
+------+                +-----+
|      |                |     |
|      |                | do1 |
|      |                |     |
|      |                +-----+
|      |                |     |
|      |                | do2 |
|      |                +-+---+
|      |                  |
|      |                  |
| A    |                  |
+------+                  |
|      |                  |
|      |                  |
|      |                  |
|      | <----------------+
|      |
|      |
|      |
+------+

1
2
3
4
5
6
7

let future = id_rpc(&my_server).and_then(|id| {
    get_row(id)
}).map(|row| {
    json::encode(row)
}).and_then(|encoded| {
    write_string(my_socket, encoded)
});

1
2
3
4

let id = id_rpc(&my_server).await;
let row = get_row(id).await;
let encoded = json::encode(row);
write_string(my_socket, encoded).await;

1
2
3
4
5

pub trait Future {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

1
2
3
4

pub enum Poll<T> {
    Ready(T),
    Pending,
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

    +------------------------------------------------------------------+
    |                                                                  |
    |    +--------------------------------------------------------+    |
    |    |                                                        |    |
    |    |   +-------------------------------------------------+  |    |
    |    |   |-------------+ +----------+ +--------------+     |  |    |
    |    |   || futureobj  | | futureobj| |  futureobj   |     |  |    |
    |    |   +-------------+ +----------+ +--------------+     |  |    |
    |    |   | 协 程  task                                      |  |    |
    |    |   +-------------------------------------------------+  |    |
    |    |                                                        |    |
    |    | 线 程                                                   |    |
    |    +--------------------------------------------------------+    |
    |                                                                  |
    |                                                                  |
    |    +--------------------------------------------------------+    |
    |    |  +--------------------------------------------------+  |    |
    |    |  |                                                  |  |    |
    |    |  |   +------------+ +-------------------------------+  |    |
    |    |  |   | futureobj  | |  futureobj      || futureobj ||  |    |
    |    |  |   +------------+ +-------------------------------+  |    |
    |    |  |  协 程 task                                      |  |    |
    |    |  +--------------------------------------------------+  |    |
    |    | 线 程                                                   |    |
    |    +--------------------------------------------------------+    |
    |                                                                  |
    | 进  程                                                            |
    +------------------------------------------------------------------+

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// async 函数，真正会返回 `Future<Output = u8>`，而不是表面看上去的 `u8`
async fn foo() -> u8 { 5 }

// async 块用法，返回 `impl Future<Output = u8>`
fn bar() -> impl Future<Output = u8> {
    // 这里 `async` 块返回  `impl Future<Output = u8>`
    async {
        let x: u8 = foo().await;
        x + 5
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

#![feature(generators, generator_trait)]
use std::ops::Generator;
use std::pin::Pin;

fn main() {
    let mut gen = || {
        yield 1;
        yield 2;
        yield 3;
        return 4;
    };
    // for _ in 0..4 {
    //     // 为了给嵌入式支持异步，多传入了一个空的unit给resume方法
    //     let c = Pin::new(&mut gen).resume(());
    //     println!("{:?}", c);
    // }


    let c = Pin::new(&mut gen).resume(());
    println!("{:?}", c);
    let c = Pin::new(&mut gen).resume(());
    println!("{:?}", c);
    let c = Pin::new(&mut gen).resume(());
    println!("{:?}", c);
    let c = Pin::new(&mut gen).resume(());
    println!("{:?}", c);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

#![allow(unused)]
#![feature(generators, generator_trait)]
use std::ops::{Generator, GeneratorState};
use std::pin::Pin;

enum __Gen {
    // (0) 初始状态
    Start,
    // (1) resume方法执行以后
    State1(State1),
    // (2) resume方法执行以后
    State2(State2),
    // (3) resume方法执行以后
    State3(State3),
    // (4) resume方法执行以后，正好完成
    Done
}


struct State1 { x: u64 }
struct State2 { x: u64 }
struct State3 { x: u64 }

impl Generator for __Gen {
    type Yield = u64;
    type Return = u64;

    fn resume(self: Pin<&mut Self>, _: ()) -> GeneratorState<u64, u64> {
        let mut_ref = self.get_mut();
        match std::mem::replace(mut_ref, __Gen::Done) {
            __Gen::Start => {
                *mut_ref = __Gen::State1(State1{x: 1});
                GeneratorState::Yielded(1)
            }
            __Gen::State1(State1{x: 1}) => {
                *mut_ref  = __Gen::State2(State2{x: 2});
                GeneratorState::Yielded(2)
        }
            __Gen::State2(State2{x: 2}) => {
                *mut_ref = __Gen::State3(State3{x: 3});
                GeneratorState::Yielded(3)
            }
            __Gen::State3(State3{x: 3}) => {
                *mut_ref  = __Gen::Done;
                GeneratorState::Complete(4)
            }
            _ => {
                panic!("generator resumed after completion")
            }
        }
    }
}

fn main(){
    let mut gen = {
        __Gen::Start
    };

    for _ in 0..4 {
        println!("{:?}", unsafe{ Pin::new(&mut gen).resume(())});
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

#![allow(unused)]
#![feature(generators, generator_trait)]
use std::pin::Pin;

use std::ops::Generator;

pub fn up_to(limit: u64) -> impl Generator<Yield = u64, Return = u64> {
    move || {
        for x in 0..limit {
            yield x;
        }
        return limit;
    }
}
fn main(){
    let a = 10;
    let mut b = up_to(a);
    unsafe {
        for _ in 0..=10{
            let c = Pin::new(&mut b).resume(());
            println!("{:?}", c);
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

#![allow(unused)]
#![feature(generators, generator_trait)]
use std::pin::Pin;

use std::ops::{Generator, GeneratorState};

pub fn up_to() -> impl Generator<Yield = u64, Return = ()> {
    move || {
        let mut x = 0;
        loop {
            x += 1;
            yield x;
        }
        return ();
    }
}
fn main(){
    let mut gen = up_to();
    unsafe {
    for _ in 0..10{
        match Pin::new(&mut gen).resume(()) {
            GeneratorState::Yielded(i) => println!("{:?}", i),
            _ => println!("Completed"),
        }
    }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

#![allow(unused)]
#![feature(generators, generator_trait)]

use std::ops::{Generator, GeneratorState};
use std::pin::Pin;

pub fn up_to(limit: u64) -> impl Generator<Yield = (), Return = Result<u64, ()>> {
    move || {
        for x in 0..limit {
            yield ();
        }
        return Ok(limit);
    }
}
fn main(){
    let limit = 2;
    let mut gen = up_to(limit);
    unsafe {
    for i in 0..=limit{
        match Pin::new(&mut gen).resume(()) {
            GeneratorState::Yielded(v) => println!("resume {:?} : Pending", i),
            GeneratorState::Complete(v) => println!("resume {:?} : Ready", i),
        }
    }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

#[lang = "from_generator"]
#[doc(hidden)]
#[unstable(feature = "gen_future", issue = "50547")]
#[rustc_const_unstable(feature = "gen_future", issue = "50547")]
#[inline]
pub const fn from_generator<T>(gen: T) -> impl Future<Output = T::Return>
where
    T: Generator<ResumeTy, Yield = ()>,
{
    #[rustc_diagnostic_item = "gen_future"]
    struct GenFuture<T: Generator<ResumeTy, Yield = ()>>(T);

    // We rely on the fact that async/await futures are immovable in order to create
    // self-referential borrows in the underlying generator.
    impl<T: Generator<ResumeTy, Yield = ()>> !Unpin for GenFuture<T> {}

    impl<T: Generator<ResumeTy, Yield = ()>> Future for GenFuture<T> {
        type Output = T::Return;
        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            // SAFETY: Safe because we're !Unpin + !Drop, and this is just a field projection.
            let gen = unsafe { Pin::map_unchecked_mut(self, |s| &mut s.0) };

            // Resume the generator, turning the `&mut Context` into a `NonNull` raw pointer. The
            // `.await` lowering will safely cast that back to a `&mut Context`.
            match gen.resume(ResumeTy(NonNull::from(cx).cast::<Context<'static>>())) {
                GeneratorState::Yielded(()) => Poll::Pending,
                GeneratorState::Complete(x) => Poll::Ready(x),
            }
        }
    }

    GenFuture(gen)
}

1
2
3
4
5

#[derive(Debug)]
pub struct CacheEntry<V> {
    pub(crate) value: V,
    pub(crate) expiration: Option<CacheExpiration>,
}

1
2
3
4

#[derive(Debug)]
pub struct CacheExpiration {
    instant: Instant,
}

1
2
3
4
5

#[derive(Debug)]
pub struct CacheEntryReadGuard<'a, V> {
    pub(crate) entry: *const CacheEntry<V>,
    pub(crate) marker: PhantomData<&'a CacheEntry<V>>,
}

1
2
3

// Stores a raw pointer to `T`, so if `T` is `Sync`, the lock guard over `T` is `Send`.
unsafe impl<V> Send for CacheEntryReadGuard<'_, V> where V: Sized + Sync {}
unsafe impl<V> Sync for CacheEntryReadGuard<'_, V> where V: Sized + Send + Sync {}

1
2
3
4

pub struct Cache<K, V> {
    store: RwLock<BTreeMap<K, CacheEntry<V>>>,
    label: String,
}

1
2
3
4
5
6
7

pub async fn monitor(&self, sample: usize, threshold: f64, frequency: Duration) {
  let mut interval = Interval::platform_new(frequency);
  loop {
    interval.as_mut().await;
    self.purge(sample, threshold).await;
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

[dependencies]
async-channel = "1.4.2"
async-executor = "1.3.0"
async-fs = "1.3.0"
async-io = "1.1.2"
async-lock = "2.3.0"
async-net = "1.4.3"
async-process = "1.0.0"
blocking = "1.0.0"
futures-lite = "1.11.0"
once_cell = "1.4.1"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

use once_cell::sync::Lazy;

/// 将异步任务加入到 task 中，交给 Global 执行器
pub fn spawn<T: Send + 'static>(future: impl Future<Output = T> + Send + 'static) -> Task<T> {
    static GLOBAL: Lazy<Executor<'_>> = Lazy::new(|| {
        let num_threads = {
            // Parse SMOL_THREADS or default to 1.
            std::env::var("SMOL_THREADS")
                .ok()
                .and_then(|s| s.parse().ok())
                .unwrap_or(1)
        };

        for n in 1..=num_threads {
            thread::Builder::new()
                .name(format!("smol-{}", n))
                .spawn(|| loop {
                    // catch_unwind 方法
                    // 可以捕获栈展开，运行panic线程存活并继续运行。使用标准库中的std::panic::catch_unwind()函数。
                    catch_unwind(|| block_on(GLOBAL.run(future::pending::<()>()))).ok();
                })
                .expect("cannot spawn executor thread");
        }

        Executor::new()
    });

    GLOBAL.spawn(future)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

/// An event on the chat server.
enum Event {
    /// A client has joined.
    Join(SocketAddr, Arc<Async<TcpStream>>),

    /// A client has left.
    Leave(SocketAddr),

    /// A client sent a message.
    Message(SocketAddr, String),
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

/// Dispatches events to clients.
async fn dispatch(receiver: Receiver<Event>) -> io::Result<()> {
    // Currently active clients.
    let mut map = HashMap::<SocketAddr, Arc<Async<TcpStream>>>::new();

    // Receive incoming events.
    while let Ok(event) = receiver.recv().await {
        // Process the event and format a message to send to clients.
        let output = match event {
            Event::Join(addr, stream) => {
                map.insert(addr, stream);
                format!("{} has joined\n", addr)
            }
            Event::Leave(addr) => {
                map.remove(&addr);
                format!("{} has left\n", addr)
            }
            Event::Message(addr, msg) => format!("{} says: {}\n", addr, msg),
        };

        // Display the event in the server process.
        print!("{}", output);

        // Send the event to all active clients.
        for stream in map.values_mut() {
            // Ignore errors because the client might disconnect at any point.
            stream.write_all(output.as_bytes()).await.ok();
        }
    }
    Ok(())
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

fn main() -> io::Result<()> {
    smol::block_on(async {
        // Create a listener for incoming client connections.
        let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 6000))?;

        // Intro messages.
        println!("Listening on {}", listener.get_ref().local_addr()?);
        println!("Start a chat client now!\n");

        // Spawn a background task that dispatches events to clients.
        // 通过 channel 实现消息传递
        let (sender, receiver) = bounded(100);
        // detach 函数
        // 分离任务，让它在后台运行
        smol::spawn(dispatch(receiver)).detach();

        loop {
            // Accept the next connection.
            let (stream, addr) = listener.accept().await?;
            let client = Arc::new(stream);
            let sender = sender.clone();

            // Spawn a background task reading messages from the client.
            smol::spawn(async move {
                // Client starts with a `Join` event.
                sender.send(Event::Join(addr, client.clone())).await.ok();

                // Read messages from the client and ignore I/O errors when the client quits.
                read_messages(sender.clone(), client).await.ok();

                // Client ends with a `Leave` event.
                sender.send(Event::Leave(addr)).await.ok();
            })
            .detach();
        }
    })
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

fn main() -> io::Result<()> {
    smol::block_on(async {
        // Connect to the server and create async stdin and stdout.
        let stream = Async::<TcpStream>::connect(([127, 0, 0, 1], 6000)).await?;
        let stdin = Unblock::new(std::io::stdin());
        let mut stdout = Unblock::new(std::io::stdout());

        // Intro messages.
        println!("Connected to {}", stream.get_ref().peer_addr()?);
        println!("My nickname: {}", stream.get_ref().local_addr()?);
        println!("Type a message and hit enter!\n");

        let reader = &stream;
        let mut writer = &stream;

        // Wait until the standard input is closed or the connection is closed.
        future::race(
            async {
                let res = io::copy(stdin, &mut writer).await;
                println!("Quit!");
                res
            },
            async {
                let res = io::copy(reader, &mut stdout).await;
                println!("Server disconnected!");
                res
            },
        )
        .await?;

        Ok(())
    })
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

#[derive(Debug)]
pub struct Timer {
    /// This timer's ID and last waker that polled it.
    ///
    /// When this field is set to `None`, this timer is not registered in the reactor.
    id_and_waker: Option<(usize, Waker)>,

    /// The next instant at which this timer fires.
    when: Instant,

    /// The period.
    period: Duration,
}

#[derive(Debug)]
pub struct Async<T> {
    /// A source registered in the reactor.
    source: Arc<Source>,

    /// The inner I/O handle.
    io: Option<T>,
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

pub(crate) struct Reactor {
    /// Portable bindings to epoll/kqueue/event ports/wepoll.
    ///
    /// This is where I/O is polled, producing I/O events.
    poller: Poller,

    /// Ticker bumped before polling.
    ///
    /// This is useful for checking what is the current "round" of `ReactorLock::react()` when
    /// synchronizing things in `Source::readable()` and `Source::writable()`. Both of those
    /// methods must make sure they don't receive stale I/O events - they only accept events from a
    /// fresh "round" of `ReactorLock::react()`.
    ticker: AtomicUsize,

    /// Registered sources.
    sources: Mutex<Slab<Arc<Source>>>,

    /// Temporary storage for I/O events when polling the reactor.
    ///
    /// Holding a lock on this event list implies the exclusive right to poll I/O.
    events: Mutex<Vec<Event>>,

    /// An ordered map of registered timers.
    ///
    /// Timers are in the order in which they fire. The `usize` in this type is a timer ID used to
    /// distinguish timers that fire at the same time. The `Waker` represents the task awaiting the
    /// timer.
    timers: Mutex<BTreeMap<(Instant, usize), Waker>>,

    /// A queue of timer operations (insert and remove).
    ///
    /// When inserting or removing a timer, we don't process it immediately - we just push it into
    /// this queue. Timers actually get processed when the queue fills up or the reactor is polled.
    timer_ops: ConcurrentQueue<TimerOp>,
}

1

let (sender, receiver) = flume::unbounded();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

pub(crate) struct Header {
    /// Current state of the task.
    ///
    /// Contains flags representing the current state and the reference count.
    pub(crate) state: AtomicUsize,

    /// The task that is blocked on the `Task` handle.
    ///
    /// This waker needs to be woken up once the task completes or is closed.
    pub(crate) awaiter: UnsafeCell<Option<Waker>>,

    /// The virtual table.
    ///
    /// In addition to the actual waker virtual table, it also contains pointers to several other
    /// methods necessary for bookkeeping the heap-allocated task.
    pub(crate) vtable: &'static TaskVTable,
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

/// Set if the task is scheduled for running.
pub(crate) const SCHEDULED: usize = 1 << 0;

/// Set if the task is running.
pub(crate) const RUNNING: usize = 1 << 1;

/// Set if the task has been completed.
pub(crate) const COMPLETED: usize = 1 << 2;

/// Set if the task is closed.
pub(crate) const CLOSED: usize = 1 << 3;

/// Set if the `Task` still exists.
pub(crate) const TASK: usize = 1 << 4;

/// Set if the `Task` is awaiting the output.
pub(crate) const AWAITER: usize = 1 << 5;

/// Set if an awaiter is being registered.
pub(crate) const REGISTERING: usize = 1 << 6;

/// Set if the awaiter is being notified.
pub(crate) const NOTIFYING: usize = 1 << 7;

/// A single reference.
pub(crate) const REFERENCE: usize = 1 << 8;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

use async_executor::Executor;
use futures_lite::future;

// Create a new executor.
let ex = Executor::new();

// Spawn a task.
let task = ex.spawn(async {
    println!("Hello world");
});

// Run the executor until the task completes.
future::block_on(ex.run(task));

1
2
3
4

#[tokio::main]
async fn main() {
    println!("Hello world");
}

1
2
3
4
5
6
7
8
9

fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("Hello world");
        })
}

1
2
3
4
5

#[proc_macro_attribute]
#[cfg(not(test))] // Work around for rust-lang/rust#62127
pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
    entry::main(args, item, true)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

#[cfg(not(test))] // Work around for rust-lang/rust#62127
pub(crate) fn main(args: TokenStream, item: TokenStream, rt_multi_thread: bool) -> TokenStream {
    // If any of the steps for this macro fail, we still want to expand to an item that is as close
    // to the expected output as possible. This helps out IDEs such that completions and other
    // related features keep working.
    let input: syn::ItemFn = match syn::parse(item.clone()) {
        Ok(it) => it,
        Err(e) => return token_stream_with_error(item, e),
    };

    let config = if input.sig.ident == "main" && !input.sig.inputs.is_empty() {
        let msg = "the main function cannot accept arguments";
        Err(syn::Error::new_spanned(&input.sig.ident, msg))
    } else {
        AttributeArgs::parse_terminated
            .parse(args)
            .and_then(|args| build_config(input.clone(), args, false, rt_multi_thread))
    };

    match config {
        Ok(config) => parse_knobs(input, false, config),
        Err(e) => token_stream_with_error(parse_knobs(input, false, DEFAULT_ERROR_CONFIG), e),
    }
}