3.3 Asynchronous green threads
In addition to the OS aware and OS-managed static threads, Deno also uses asynchronous green threads to carry out async operations. There is a subtle difference between static threads and green threads.
The static threads are allocated as a normal thread. The operating system is aware of the static threads, therefore it manages their lifecycle, from start to end. These threads get scheduled by the operating system.
On the other hand, the asynchronous green threads are basically tokio tasks. The lifecycle of these threads is managed by the tokio runtime. This means that the green threads get scheduled by the tokio runtime. The operating system is unaware of tokio tasks or green threads. This makes tokio tasks lightning fast. They are very efficient.
A task or a green thread is a lightweight, non-blocking unit of execution. A task is similar to an OS thread, but rather than being managed by the OS scheduler, it is managed by the Tokio runtime. Another name for this general pattern is green threads.
There are certain key characteristics of tokio tasks:
Tasks are lightweight. Because tasks are scheduled by the Tokio runtime rather than the operating system, creating new tasks or switching between tasks does not require a context switch and has fairly low overhead. Creating, running, and destroying large numbers of tasks is quite cheap, especially compared to OS threads. This makes them very efficient.
Tasks are scheduled cooperatively. Most operating systems implement preemptive multitasking. This is a scheduling technique where the operating system allows each thread to run for a period of time and then preempts it, temporarily pausing that thread and switching to another. Tasks, on the other hand, implement cooperative multitasking. In cooperative multitasking, a task is allowed to run until it yields, indicating to the Tokio runtime's scheduler that it cannot currently continue executing. When a task yields, the Tokio runtime switches to executing the next task. Or in other words, tokio tasks don't get preempted till they yield.
Tasks are non-blocking. Typically, when an OS thread performs I/O or must synchronize with another thread, it blocks, allowing the OS to schedule another thread. When a task cannot continue executing, it must yield instead, allowing the Tokio runtime to schedule another task. Tasks should generally not perform system calls or other operations that could block a thread, as this would prevent other tasks running on the same thread from executing as well. Remember these tasks aren't real threads. They are like virtual threads. They are running in the context of the main thread. So, if a green thread blocks on something, it has blocked everything. Instead, tasks should use APIs for running blocking operations in an asynchronous context. Tokio has many equivalent async APIs.
A tokio task or a green thread needs to be spawned. There are three different ways to spawn a tokio task.
This function is an async equivalent to the standard library's thread::spawn. It takes an async block or other future and creates a new task to run that work concurrently.
Spawning a task enables the task to execute concurrently to other tasks. The spawned task may execute on the current thread, or it may be sent to a different thread to be executed. The specifics depend on the current Runtime configuration.
There is no guarantee that a spawned task will execute to completion. When a runtime gets shutdown, all outstanding tasks are dropped, regardless of the lifecycle of that task.
use tokio::task;
task::spawn(async {
// do async work
});
Task spawn() takes an async block to execute. It returns a join handle which can be awaited till the spawned task completes.
Here is an example of task spawning from tokio documentation:
use tokio::net::{TcpListener, TcpStream};
use std::io;
async fn process(socket: TcpStream) {
// ...
}
#[tokio::main]
async fn main() -> io::Result<()> {
let mut listener = TcpListener::bind("127.0.0.1:8080").await?;
loop {
let (socket, _) = listener.accept().await?;
tokio::spawn(async move {
// Process each socket concurrently.
process(socket).await
});
}
}
This is a special way to create tasks that are going to block the execution. A blocking operation performed in a task running on a thread that is also running other tasks would block the entire thread, preventing other tasks from running. The task::spawn_blocking function is similar to the task::spawn function discussed in the previous section, but rather than spawning a non-blocking future on the Tokio runtime, it instead spawns a blocking function on a dedicated thread pool for blocking tasks.
In general, issuing a blocking call or performing a lot of computing without yielding is not okay, as it may prevent the executor from driving other futures forward. A closure that is run through this method will instead be run on a dedicated thread pool for such blocking tasks without holding up the main futures executor.
use tokio::task;
task::spawn_blocking(|| {
// do sync work
});
Blocking tasks runs on a dedicated thread pool. These tasks are useful to do sync work or computation heavy work.
Here is an example from tokio documentation:
use tokio::task;
let res = task::spawn_blocking(move || {
// do some compute-heavy work or call synchronous code
"done computing"
}).await?;
assert_eq!(res, "done computing");
This function transitions the current worker thread to a blocking thread, moving other tasks running on that thread to another worker thread. This can improve performance by avoiding context switches.
Block in place is special as it converts the current thread to a blocking mode. Tokio's scheduler takes care of moving other tasks out of this green thread.
use tokio::task;
task::block_in_place(|| {
// do sync work
});
The primary need from Deno is to run code asynchronously. There are two ways Deno achieves it:
- Async functions
- Tokio comes packaged with several asynchronous functions which are equivalent of the blocking synchronous functions
- Tasks
- For functions not provided by tokio, Deno uses tasks to run them asynchronously
Deno doesn't use task::spawn which creates a new task. The reason likely is performance. A simple spawn creates a new green thread which could be a bit of overhead to execute a simple async code.
Instead of spawn, Deno uses spawn_blocking to run async ops, and that too is used mostly for file system related operations. For other async tasks, Deno directly uses the async APIs available either through tokio or from other third-party libraries.
Let's go over one example of each from Deno.
First, let's take an example of an async op that doesn't need an explicit task to be created. There is no need to spawn a task because Deno directly uses async functions.
Consider the case of accepting a TCP connection. This is usually a blocking call. Deno implements TCP accept via a low-level op called op_accept. This op accepts the incoming TCP connections. As tokio already supports async TCP and UDP, there is no need to do anything special.
async fn op_accept(
state: Rc<RefCell<OpState>>,
args: Value,
bufs: BufVec,
) -> Result<Value, AnyError> {
let args: AcceptArgs = serde_json::from_value(args)?;
match args.transport.as_str() {
"tcp" => accept_tcp(state, args, bufs).await,
#[cfg(unix)]
"unix" => net_unix::accept_unix(state, args, bufs).await,
_ => Err(generic_error(format!(
"Unsupported transport protocol {}",
args.transport
))),
}
}
async fn accept_tcp(
state: Rc<RefCell<OpState>>,
args: AcceptArgs,
_zero_copy: BufVec,
) -> Result<Value, AnyError> {
let rid = args.rid as u32;
let accept_fut = poll_fn(|cx| {
let mut state = state.borrow_mut();
let listener_resource = state
.resource_table
.get_mut::<TcpListenerResource>(rid)
.ok_or_else(|| bad_resource("Listener has been closed"))?;
let listener = &mut listener_resource.listener;
match listener.poll_accept(cx).map_err(AnyError::from) {
Poll::Ready(Ok((stream, addr))) => {
listener_resource.untrack_task();
Poll::Ready(Ok((stream, addr)))
}
Poll::Pending => {
listener_resource.track_task(cx)?;
Poll::Pending
}
Poll::Ready(Err(e)) => {
listener_resource.untrack_task();
Poll::Ready(Err(e))
}
}
});
let (tcp_stream, _socket_addr) = accept_fut.await?;
let local_addr = tcp_stream.local_addr()?;
let remote_addr = tcp_stream.peer_addr()?;
// -- CODE OMITTED --
}))
}
In op_accept(), Deno makes a call to the accept_tcp() which does an asynchronously wait for new connections.
As mentioned earlier, Deno directly uses asynchronous functions offered by tokio.
Second, let's take an example of some sync operation that Deno likes to run as async operations. A good example is file system operations. In Deno, only async file system ops are spawned as blocking.
Here are some of the async file system ops:
- Open
- Seek
- Fstat
- Mkdir
- Chmod
- Chown
- Copy file
- Read dir
- .... and many more (the list is quite long)
Only the asynchronous file system ops get spawned as a blocking task. This means that these ops will likely get scheduled on tokio's dedicated thread pool.
Here are the sync and async implementation of a very simple op utime:
fn op_utime_sync(
state: &mut OpState,
args: Value,
_zero_copy: &mut [ZeroCopyBuf],
) -> Result<Value, AnyError> {
super::check_unstable(state, "Deno.utime");
let args: UtimeArgs = serde_json::from_value(args)?;
let path = PathBuf::from(&args.path);
let atime = filetime::FileTime::from_unix_time(args.atime.0, args.atime.1);
let mtime = filetime::FileTime::from_unix_time(args.mtime.0, args.mtime.1);
state.borrow::<Permissions>().check_write(&path)?;
filetime::set_file_times(path, atime, mtime)?;
Ok(json!({}))
}
async fn op_utime_async(
state: Rc<RefCell<OpState>>,
args: Value,
_zero_copy: BufVec,
) -> Result<Value, AnyError> {
let state = state.borrow();
super::check_unstable(&state, "Deno.utime");
let args: UtimeArgs = serde_json::from_value(args)?;
let path = PathBuf::from(&args.path);
let atime = filetime::FileTime::from_unix_time(args.atime.0, args.atime.1);
let mtime = filetime::FileTime::from_unix_time(args.mtime.0, args.mtime.1);
state.borrow::<Permissions>().check_write(&path)?;
tokio::task::spawn_blocking(move || {
filetime::set_file_times(path, atime, mtime)?;
Ok(json!({}))
})
.await
.unwrap()
}
These two functions are almost the same. They call the same set_file_times(). However, there is a big difference.
In the sync variant, the call to set file time blocks the current thread till the set_file_times() finishes its work.
In the async variant, a blocking task is spawned which makes the call to set file time. This doesn't block the thread as the operation runs on a dedicated thread pool present in tokio. There is an await at the end of spawn_blocking. This await would wait for the task to finish before proceeding further.
// Sync
filetime::set_file_times(path, atime, mtime)?;
// Async
task::spawn_blocking( { ...... filetime::set_file_times.......} ).await
That was all about the asynchronous capabilities through ready APIs or tasks.