feat: migrate from async-std to tokio

depends on
- [ ] chatmail/async-smtp#52
- [ ] chatmail/async-imap#21

Author: dignifiedquire

Cargo.lock

@@ -19,12 +19,12 @@ deltachat_derive = { path = "./deltachat_derive" }
 
 ansi_term = { version = "0.12.1", optional = true }
 anyhow = "1"
-async-imap = { git = "https://github.com/async-email/async-imap" }
-async-native-tls = { version = "0.3" }
-async-smtp = { git = "https://github.com/async-email/async-smtp", branch="master", default-features=false, features = ["smtp-transport", "socks5"] }
-async-std-resolver = "0.21"
-async-std = { version = "1" }
-async-tar = { version = "0.4", default-features=false }
+async-imap = { git = "https://github.com/async-email/async-imap", branch = "runtimes", default-features = false, features = ["runtime-tokio"] }
+async-native-tls = { version = "0.4", default-features = false, features = ["runtime-tokio"] }
+async-smtp = { git = "https://github.com/async-email/async-smtp", branch="tokio", default-features = false, features = ["smtp-transport", "socks5", "runtime-tokio"] }
+trust-dns-resolver = "0.21"
+tokio = { version = "1", features = ["full"] }
+tokio-tar = { version = "0.3" } # TODO: integrate tokio into async-tar
 backtrace = "0.3"
 base64 = "0.13"
 bitflags = "1.3"
@@ -65,22 +65,24 @@ sha2 = "0.10"
 smallvec = "1"
 strum = "0.24"
 strum_macros = "0.24"
-surf = { version = "2.3", default-features = false, features = ["h1-client"] }
 thiserror = "1"
 toml = "0.5"
 url = "2"
 uuid = { version = "1", features = ["serde", "v4"] }
-fast-socks5 = "0.4"
+fast-socks5 = "0.8"
 humansize = "1"
 qrcodegen = "1.7.0"
 tagger = "4.3.3"
 textwrap = "0.15.0"
 zip = { version = "0.6.2", default-features = false, features = ["deflate"] }
+async-channel = "1.6.1"
+futures-lite = "1.12.0"
+tokio-stream = { version = "0.1.9", features = ["fs"] }
+reqwest = { version = "0.11.11", features = ["json"] }
 
 [dev-dependencies]
 ansi_term = "0.12.0"
-async-std = { version = "1", features = ["unstable", "attributes"] }
-criterion = { version = "0.3.4", features = ["async_std"] }
+criterion = { version = "0.3.4", features = ["async_tokio"] }
 futures-lite = "1.12"
 log = "0.4"
 pretty_env_logger = "0.4"
@@ -132,5 +134,10 @@ harness = false
 default = ["vendored"]
 internals = []
 repl = ["internals", "rustyline", "log", "pretty_env_logger", "ansi_term", "dirs"]
-vendored = ["async-native-tls/vendored", "async-smtp/native-tls-vendored", "rusqlite/bundled-sqlcipher-vendored-openssl"]
+vendored = [
+  "async-native-tls/vendored",
+  "async-smtp/native-tls-vendored",
+  "rusqlite/bundled-sqlcipher-vendored-openssl",
+  "reqwest/native-tls-vendored"
+]
 nightly = ["pgp/nightly"]

Cargo.toml

@@ -1,4 +1,3 @@
-use async_std::task::block_on;
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use deltachat::contact::Contact;
 use deltachat::context::Context;
@@ -27,12 +26,16 @@ async fn address_book_benchmark(n: u32, read_count: u32) {
 }
 
 fn criterion_benchmark(c: &mut Criterion) {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
     c.bench_function("create 500 contacts", |b| {
-        b.iter(|| block_on(async { address_book_benchmark(black_box(500), black_box(0)).await }))
+        b.to_async(&rt)
+            .iter(|| async { address_book_benchmark(black_box(500), black_box(0)).await })
     });
 
     c.bench_function("create 100 contacts and read it 1000 times", |b| {
-        b.iter(|| block_on(async { address_book_benchmark(black_box(100), black_box(1000)).await }))
+        b.to_async(&rt)
+            .iter(|| async { address_book_benchmark(black_box(100), black_box(1000)).await })
     });
 }
 

benches/contacts.rs

@@ -1,7 +1,6 @@
-use async_std::path::PathBuf;
-use async_std::task::block_on;
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use deltachat::accounts::Accounts;
+use std::path::PathBuf;
 use tempfile::tempdir;
 
 async fn create_accounts(n: u32) {
@@ -18,7 +17,8 @@ async fn create_accounts(n: u32) {
 
 fn criterion_benchmark(c: &mut Criterion) {
     c.bench_function("create 1 account", |b| {
-        b.iter(|| block_on(async { create_accounts(black_box(1)).await }))
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        b.to_async(&rt).iter(|| create_accounts(black_box(1)))
     });
 }
 

benches/create_account.rs

@@ -1,6 +1,5 @@
-use async_std::path::Path;
+use std::path::Path;
 
-use criterion::async_executor::AsyncStdExecutor;
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 
 use deltachat::chat::{self, ChatId};
@@ -23,7 +22,9 @@ fn criterion_benchmark(c: &mut Criterion) {
     // To enable this benchmark, set `DELTACHAT_BENCHMARK_DATABASE` to some large database with many
     // messages, such as your primary account.
     if let Ok(path) = std::env::var("DELTACHAT_BENCHMARK_DATABASE") {
-        let chats: Vec<_> = async_std::task::block_on(async {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+
+        let chats: Vec<_> = rt.block_on(async {
             let context = Context::new((&path).into(), 100, Events::new())
                 .await
                 .unwrap();
@@ -33,7 +34,7 @@ fn criterion_benchmark(c: &mut Criterion) {
         });
 
         c.bench_function("chat::get_chat_msgs (load messages from 10 chats)", |b| {
-            b.to_async(AsyncStdExecutor)
+            b.to_async(&rt)
                 .iter(|| get_chat_msgs_benchmark(black_box(path.as_ref()), black_box(&chats)))
         });
     } else {

benches/get_chat_msgs.rs

@@ -1,4 +1,3 @@
-use criterion::async_executor::AsyncStdExecutor;
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 
 use deltachat::chatlist::Chatlist;
@@ -13,11 +12,11 @@ fn criterion_benchmark(c: &mut Criterion) {
     // To enable this benchmark, set `DELTACHAT_BENCHMARK_DATABASE` to some large database with many
     // messages, such as your primary account.
     if let Ok(path) = std::env::var("DELTACHAT_BENCHMARK_DATABASE") {
-        let context = async_std::task::block_on(async {
-            Context::new(path.into(), 100, Events::new()).await.unwrap()
-        });
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let context =
+            rt.block_on(async { Context::new(path.into(), 100, Events::new()).await.unwrap() });
         c.bench_function("chatlist:try_load (Get Chatlist)", |b| {
-            b.to_async(AsyncStdExecutor)
+            b.to_async(&rt)
                 .iter(|| get_chat_list_benchmark(black_box(&context)))
         });
     } else {

benches/get_chatlist.rs

@@ -1,8 +1,6 @@
-use async_std::{path::PathBuf, task::block_on};
-use criterion::{
-    async_executor::AsyncStdExecutor, black_box, criterion_group, criterion_main, BatchSize,
-    Criterion,
-};
+use std::path::PathBuf;
+
+use criterion::{black_box, criterion_group, criterion_main, BatchSize, Criterion};
 use deltachat::{
     config::Config,
     context::Context,
@@ -51,7 +49,7 @@ async fn create_context() -> Context {
         .unwrap()
         .join("delta-chat-backup.tar")
         .into();
-    if backup.exists().await {
+    if backup.exists() {
         println!("Importing backup");
         imex(&context, ImexMode::ImportBackup, &backup, None)
             .await
@@ -74,8 +72,10 @@ async fn create_context() -> Context {
 fn criterion_benchmark(c: &mut Criterion) {
     let mut group = c.benchmark_group("Receive messages");
     group.bench_function("Receive 100 simple text msgs", |b| {
-        b.to_async(AsyncStdExecutor).iter_batched(
-            || block_on(create_context()),
+        let rt = tokio::runtime::Runtime::new().unwrap();
+
+        b.to_async(&rt).iter_batched(
+            || rt.block_on(create_context()),
             |context| recv_all_emails(black_box(context)),
             BatchSize::LargeInput,
         );

benches/receive_emails.rs

@@ -1,4 +1,3 @@
-use async_std::task::block_on;
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use deltachat::context::Context;
 use deltachat::Events;
@@ -20,8 +19,10 @@ fn criterion_benchmark(c: &mut Criterion) {
     // To enable this benchmark, set `DELTACHAT_BENCHMARK_DATABASE` to some large database with many
     // messages, such as your primary account.
     if let Ok(path) = std::env::var("DELTACHAT_BENCHMARK_DATABASE") {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+
         c.bench_function("search hello", |b| {
-            b.iter(|| block_on(async { search_benchmark(black_box(&path)).await }))
+            b.to_async(&rt).iter(|| search_benchmark(black_box(&path)))
         });
     } else {
         println!("env var not set: DELTACHAT_BENCHMARK_DATABASE");

benches/search_msgs.rs

@@ -20,10 +20,11 @@ libc = "0.2"
 human-panic = "1"
 num-traits = "0.2"
 serde_json = "1.0"
-async-std = "1"
+tokio = { version = "1", features = ["full"] }
 anyhow = "1"
 thiserror = "1"
 rand = "0.7"
+lazy_static = "1.4.0"
 
 [features]
 default = ["vendored"]

deltachat-ffi/Cargo.toml

@@ -15,17 +15,19 @@ extern crate human_panic;
 use std::collections::BTreeMap;
 use std::convert::TryFrom;
 use std::fmt::Write;
+use std::future::Future;
 use std::ops::Deref;
 use std::ptr;
 use std::str::FromStr;
 use std::time::{Duration, SystemTime};
 
 use anyhow::Context as _;
-use async_std::sync::RwLock;
-use async_std::task::{block_on, spawn};
 use deltachat::qr_code_generator::get_securejoin_qr_svg;
 use num_traits::{FromPrimitive, ToPrimitive};
 use rand::Rng;
+use tokio::runtime::Runtime;
+use tokio::sync::RwLock;
+use tokio::task::spawn;
 
 use deltachat::chat::{ChatId, ChatVisibility, MuteDuration, ProtectionStatus};
 use deltachat::constants::DC_MSG_ID_LAST_SPECIAL;
@@ -61,6 +63,17 @@ use deltachat::chatlist::Chatlist;
 /// Struct representing the deltachat context.
 pub type dc_context_t = Context;
 
+lazy_static::lazy_static! {
+    static ref RT: Runtime = Runtime::new().expect("unable to create tokio runtime");
+}
+
+fn block_on<F, T>(fut: F) -> T
+where
+    F: Future<Output = T>,
+{
+    RT.block_on(fut)
+}
+
 #[no_mangle]
 pub unsafe extern "C" fn dc_context_new(
     _os_name: *const libc::c_char,
@@ -4298,7 +4311,7 @@ pub unsafe extern "C" fn dc_accounts_migrate_account(
     block_on(async move {
         let mut accounts = accounts.write().await;
         match accounts
-            .migrate_account(async_std::path::PathBuf::from(dbfile))
+            .migrate_account(std::path::PathBuf::from(dbfile))
             .await
         {
             Ok(id) => id,

deltachat-ffi/src/lib.rs

@@ -1,9 +1,9 @@
 extern crate dirs;
 
+use std::path::Path;
 use std::str::FromStr;
 
 use anyhow::{bail, ensure, Result};
-use async_std::path::Path;
 use deltachat::chat::{
     self, Chat, ChatId, ChatItem, ChatVisibility, MuteDuration, ProtectionStatus,
 };
@@ -492,7 +492,7 @@ pub async fn cmdline(context: Context, line: &str, chat_id: &mut ChatId) -> Resu
             let setup_code = create_setup_code(&context);
             let file_name = blobdir.join("autocrypt-setup-message.html");
             let file_content = render_setup_file(&context, &setup_code).await?;
-            async_std::fs::write(&file_name, file_content).await?;
+            tokio::fs::write(&file_name, file_content).await?;
             println!(
                 "Setup message written to: {}\nSetup code: {}",
                 file_name.display(),

examples/repl/cmdline.rs

@@ -9,11 +9,11 @@ extern crate deltachat;
 
 use std::borrow::Cow::{self, Borrowed, Owned};
 use std::io::{self, Write};
+use std::path::Path;
 use std::process::Command;
 
 use ansi_term::Color;
 use anyhow::{bail, Error};
-use async_std::path::Path;
 use deltachat::chat::ChatId;
 use deltachat::config;
 use deltachat::context::*;
@@ -30,6 +30,7 @@ use rustyline::validate::Validator;
 use rustyline::{
     Cmd, CompletionType, Config, Context as RustyContext, EditMode, Editor, Helper, KeyEvent,
 };
+use tokio::runtime::Handle;
 
 mod cmdline;
 use self::cmdline::*;
@@ -301,7 +302,7 @@ async fn start(args: Vec<String>) -> Result<(), Error> {
     let context = Context::new(Path::new(&args[1]).to_path_buf(), 0, Events::new()).await?;
 
     let events = context.get_event_emitter();
-    async_std::task::spawn(async move {
+    tokio::task::spawn(async move {
         while let Some(event) = events.recv().await {
             receive_event(event.typ);
         }
@@ -316,8 +317,8 @@ async fn start(args: Vec<String>) -> Result<(), Error> {
         .output_stream(OutputStreamType::Stdout)
         .build();
     let mut selected_chat = ChatId::default();
-    let (reader_s, reader_r) = async_std::channel::bounded(100);
-    let input_loop = async_std::task::spawn_blocking(move || {
+    let (reader_s, reader_r) = async_channel::bounded(100);
+    let input_loop = tokio::task::spawn_blocking(move || {
         let h = DcHelper {
             completer: FilenameCompleter::new(),
             highlighter: MatchingBracketHighlighter::new(),
@@ -339,7 +340,7 @@ async fn start(args: Vec<String>) -> Result<(), Error> {
                 Ok(line) => {
                     // TODO: ignore "set mail_pw"
                     rl.add_history_entry(line.as_str());
-                    async_std::task::block_on(reader_s.send(line)).unwrap();
+                    Handle::current().block_on(reader_s.send(line)).unwrap();
                 }
                 Err(ReadlineError::Interrupted) | Err(ReadlineError::Eof) => {
                     println!("Exiting...");
@@ -367,7 +368,7 @@ async fn start(args: Vec<String>) -> Result<(), Error> {
         }
     }
     context.stop_io().await;
-    input_loop.await?;
+    input_loop.await??;
 
     Ok(())
 }
@@ -458,11 +459,12 @@ async fn handle_cmd(
     Ok(ExitResult::Continue)
 }
 
-fn main() -> Result<(), Error> {
+#[tokio::main]
+async fn main() -> Result<(), Error> {
     let _ = pretty_env_logger::try_init();
 
     let args = std::env::args().collect();
-    async_std::task::block_on(async move { start(args).await })?;
+    start(args).await?;
 
     Ok(())
 }

examples/repl/main.rs

@@ -29,7 +29,7 @@ fn cb(event: EventType) {
 }
 
 /// Run with `RUST_LOG=simple=info cargo run --release --example simple --features repl -- email pw`.
-#[async_std::main]
+#[tokio::main]
 async fn main() {
     pretty_env_logger::try_init_timed().ok();
 
@@ -43,7 +43,7 @@ async fn main() {
     log::info!("info: {:#?}", info);
 
     let events = ctx.get_event_emitter();
-    let events_spawn = async_std::task::spawn(async move {
+    let events_spawn = tokio::task::spawn(async move {
         while let Some(event) = events.recv().await {
             cb(event.typ);
         }
@@ -80,7 +80,7 @@ async fn main() {
     }
 
     // wait for the message to be sent out
-    async_std::task::sleep(std::time::Duration::from_secs(1)).await;
+    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 
     log::info!("fetching chats..");
     let chats = Chatlist::try_load(&ctx, 0, None, None).await.unwrap();