Introduce config caching

Don't count the account creation in the receive emails benchmark

Use Criterion's async support

See https://bheisler.github.io/criterion.rs/book/user_guide/benchmarking_async.html

Cargo.lock

@@ -847,11 +847,13 @@ version = "0.3.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1604dafd25fba2fe2d5895a9da139f8dc9b319a5fe5354ca137cbbce4e178d10"
 dependencies = [
+ "async-std",
  "atty",
  "cast",
  "clap",
  "criterion-plot",
  "csv",
+ "futures",
  "itertools",
  "lazy_static",
  "num-traits",

Cargo.toml

@@ -79,7 +79,7 @@ humansize = "1"
 [dev-dependencies]
 ansi_term = "0.12.0"
 async-std = { version = "1", features = ["unstable", "attributes"] }
-criterion = "0.3"
+criterion = { version = "0.3.4", features = ["async_std"] }
 futures-lite = "1.12"
 log = "0.4"
 pretty_assertions = "1.0"
@@ -116,6 +116,10 @@ harness = false
 name = "search_msgs"
 harness = false
 
+[[bench]]
+name = "receive_emails"
+harness = false
+
 [features]
 default = ["vendored"]
 internals = []

benches/create_account.rs

@@ -8,7 +8,7 @@ async fn create_accounts(n: u32) {
     let dir = tempdir().unwrap();
     let p: PathBuf = dir.path().join("accounts").into();
 
-    let accounts = Accounts::new("my_os".into(), p.clone()).await.unwrap();
+    let mut accounts = Accounts::new("my_os".into(), p.clone()).await.unwrap();
 
     for expected_id in 2..n {
         let id = accounts.add_account().await.unwrap();

benches/receive_emails.rs

@@ -0,0 +1,117 @@
+use std::convert::TryInto;
+
+use async_std::{path::PathBuf, task::block_on};
+use criterion::{
+    async_executor::AsyncStdExecutor, black_box, criterion_group, criterion_main, BatchSize,
+    BenchmarkId, Criterion,
+};
+use deltachat::{
+    config::Config,
+    context::Context,
+    dc_receive_imf::dc_receive_imf,
+    imex::{imex, ImexMode},
+};
+use tempfile::tempdir;
+
+async fn recv_emails(context: Context, emails: &[&[u8]]) -> Context {
+    for (i, bytes) in emails.iter().enumerate() {
+        dc_receive_imf(
+            &context,
+            bytes,
+            "INBOX",
+            black_box(i.try_into().unwrap()),
+            false,
+        )
+        .await
+        .unwrap();
+    }
+    context
+}
+
+async fn recv_all_emails(mut context: Context, needs_move_enabled: bool) -> Context {
+    context.disable_needs_move = !needs_move_enabled;
+
+    for i in 0..100 {
+        let imf_raw = format!(
+            "Subject: Benchmark
+Message-ID: Mr.OssSYnOFkhR.{i}@testrun.org
+Date: Sat, 07 Dec 2019 19:00:27 +0000
+To: alice@example.com
+From: sender@testrun.org
+Chat-Version: 1.0
+Chat-Disposition-Notification-To: sender@testrun.org
+Chat-User-Avatar: 0
+In-Reply-To: Mr.OssSYnOFkhR.{i_dec}@testrun.org
+MIME-Version: 1.0
+
+Content-Type: text/plain; charset=utf-8; format=flowed; delsp=no
+
+Hello {i}",
+            i = i,
+            i_dec = i - 1,
+        );
+        dc_receive_imf(
+            &context,
+            imf_raw.as_bytes(),
+            "INBOX",
+            black_box(i.try_into().unwrap()),
+            false,
+        )
+        .await
+        .unwrap();
+    }
+    context
+}
+
+async fn create_context() -> Context {
+    let dir = tempdir().unwrap();
+    let dbfile = dir.path().join("db.sqlite");
+    let id = 100;
+    let context = Context::new("FakeOS".into(), dbfile.into(), id)
+        .await
+        .unwrap();
+
+    let backup: PathBuf = std::env::current_dir()
+        .unwrap()
+        .join("delta-chat-backup.tar")
+        .into();
+    if backup.exists().await {
+        println!("Importing backup");
+        imex(&context, ImexMode::ImportBackup, &backup)
+            .await
+            .unwrap();
+    }
+
+    let addr = "alice@example.com";
+    context.set_config(Config::Addr, Some(addr)).await.unwrap();
+    context
+        .set_config(Config::ConfiguredAddr, Some(addr))
+        .await
+        .unwrap();
+    context
+        .set_config(Config::Configured, Some("1"))
+        .await
+        .unwrap();
+    context
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("from_elem");
+    for needs_move_enabled in [false, true] {
+        group.bench_with_input(
+            BenchmarkId::new("Receive many messages", needs_move_enabled),
+            &needs_move_enabled,
+            |b, needs_move_enabled| {
+                b.to_async(AsyncStdExecutor).iter_batched(
+                    || block_on(create_context()),
+                    |context| recv_all_emails(black_box(context), *needs_move_enabled),
+                    BatchSize::LargeInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);

src/context.rs

@@ -30,6 +30,7 @@ use crate::sql::Sql;
 #[derive(Clone, Debug)]
 pub struct Context {
     pub(crate) inner: Arc<InnerContext>,
+    pub disable_needs_move: bool, // TODO just added for some extra profiling info
 }
 
 impl Deref for Context {
@@ -157,6 +158,7 @@ impl Context {
 
         let ctx = Context {
             inner: Arc::new(inner),
+            disable_needs_move: false,
         };
         ctx.sql.open(&ctx, &ctx.dbfile, false).await?;
 

src/dc_receive_imf.rs

@@ -193,7 +193,7 @@ pub(crate) async fn dc_receive_imf_inner(
     }
 
     // Add parts
-    let chat_id = add_parts(
+    let (chat_id, msgs) = add_parts(
         context,
         &mut mime_parser,
         imf_raw,
@@ -323,7 +323,11 @@ pub(crate) async fn dc_receive_imf_inner(
                 )
                 .await?;
             }
-        } else if insert_msg_id
+        } else if msgs
+            .first()
+            .unwrap() // TODO unwrap() (should be safe though)
+            .clone() // TODO unnecessary clone
+            .into_lazy()
             .needs_move(context, server_folder.as_ref())
             .await
             .unwrap_or_default()
@@ -439,7 +443,7 @@ async fn add_parts(
     create_event_to_send: &mut Option<CreateEvent>,
     fetching_existing_messages: bool,
     prevent_rename: bool,
-) -> Result<ChatId> {
+) -> Result<(ChatId, Vec<Message>)> {
     let mut chat_id = None;
     let mut chat_id_blocked = Blocked::Not;
     let mut incoming_origin = incoming_origin;
@@ -513,7 +517,7 @@ async fn add_parts(
                 }
                 Err(err) => {
                     warn!(context, "Error in Secure-Join message handling: {}", err);
-                    return Ok(DC_CHAT_ID_TRASH);
+                    return Ok((DC_CHAT_ID_TRASH, Vec::new()));
                 }
             }
         } else {
@@ -740,7 +744,7 @@ async fn add_parts(
                 }
                 Err(err) => {
                     warn!(context, "Error in Secure-Join watching: {}", err);
-                    return Ok(DC_CHAT_ID_TRASH);
+                    return Ok((DC_CHAT_ID_TRASH, Vec::new()));
                 }
             }
         } else if mime_parser.sync_items.is_some() && self_sent {
@@ -1005,7 +1009,7 @@ async fn add_parts(
                                 sort_timestamp,
                             )
                             .await?;
-                            return Ok(chat_id); // do not return an error as this would result in retrying the message
+                            return Ok((chat_id, Vec::new())); // do not return an error as this would result in retrying the message
                         }
                     }
                     set_better_msg(
@@ -1076,6 +1080,7 @@ async fn add_parts(
     let mut ids = Vec::with_capacity(parts.len());
 
     let conn = context.sql.get_conn().await?;
+    let mut msgs = Vec::new();
 
     for part in &mut parts {
         let mut txt_raw = "".to_string();
@@ -1133,6 +1138,53 @@ INSERT INTO msgs
         // also change `MsgId::trash()` and `delete_expired_messages()`
         let trash = chat_id.is_trash();
 
+        let msg = Message {
+            id: MsgId::new(0),
+            rfc724_mid: rfc724_mid.to_string(),
+            server_uid: server_uid,
+            chat_id,
+            from_id: if trash { 0 } else { from_id },
+            to_id: if trash { 0 } else { to_id },
+            timestamp_sort: sort_timestamp,
+            timestamp_sent: sent_timestamp,
+            timestamp_rcvd: rcvd_timestamp,
+            viewtype: part.typ,
+            state,
+            is_dc_message,
+            text: if trash {
+                None
+            } else {
+                Some(part.msg.to_string())
+            },
+            subject: if trash {
+                "".to_string()
+            } else {
+                subject.to_string()
+            },
+            param: if trash {
+                Params::new()
+            } else {
+                part.param.clone()
+            },
+            in_reply_to: Some(mime_in_reply_to.to_string()), // TODO be careful with Some("") and None
+            mime_modified,
+            error: part.error.clone(),
+            ephemeral_timer,
+            ephemeral_timestamp,
+            download_state: if is_partial_download.is_some() {
+                DownloadState::Available
+            } else {
+                DownloadState::Done
+            },
+            hidden: false,
+            chat_blocked: chat_id_blocked, // TODO not sure if correct
+            location_id: 0,
+            server_folder: Some(server_folder.to_string()),
+        };
+        // TODO mabye reuse some of the fields from above in the query below to avoid duplicate code
+
+        msgs.push(msg);
+
         stmt.execute(paramsv![
             rfc724_mid,
             server_folder,
@@ -1223,7 +1275,7 @@ INSERT INTO msgs
         }
     }
 
-    Ok(chat_id)
+    Ok((chat_id, msgs))
 }
 
 /// Saves attached locations to the database.

src/imap.rs

@@ -1598,6 +1598,7 @@ async fn precheck_imf(
 
             if delete_server_after != Some(0) {
                 if msg_id
+                    .to_lazy()
                     .needs_move(context, server_folder)
                     .await
                     .unwrap_or_default()

src/job.rs

@@ -561,11 +561,16 @@ impl Job {
         }
 
         let msg = job_try!(Message::load_from_db(context, MsgId::new(self.foreign_id)).await);
-        let server_folder = &job_try!(msg
+        let server_folder = job_try!(msg
             .server_folder
+            .as_ref()
             .context("Can't move message out of folder if we don't know the current folder"));
 
-        let move_res = msg.id.needs_move(context, server_folder).await;
+        let move_res = msg
+            .clone() // TODO avoid clone()?
+            .into_lazy()
+            .needs_move(context, &server_folder)
+            .await;
         let dest_folder = match move_res {
             Err(e) => {
                 warn!(context, "could not load dest folder: {}", e);
@@ -589,7 +594,7 @@ impl Job {
 
         if let Some(dest_folder) = dest_folder {
             match imap
-                .mv(context, server_folder, msg.server_uid, &dest_folder)
+                .mv(context, &server_folder, msg.server_uid, &dest_folder)
                 .await
             {
                 ImapActionResult::RetryLater => Status::RetryLater,

src/message.rs

@@ -6,6 +6,7 @@ use std::convert::TryInto;
 use anyhow::{ensure, format_err, Context as _, Result};
 use async_std::path::{Path, PathBuf};
 use deltachat_derive::{FromSql, ToSql};
+use once_cell::sync::OnceCell;
 use rusqlite::types::ValueRef;
 use serde::{Deserialize, Serialize};
 
@@ -53,6 +54,13 @@ impl MsgId {
         MsgId(0)
     }
 
+    pub fn to_lazy(&self) -> LazyMsg {
+        LazyMsg {
+            id: *self,
+            message: OnceCell::new(),
+        }
+    }
+
     /// Whether the message ID signifies a special message.
     ///
     /// This kind of message ID can not be used for real messages.
@@ -83,65 +91,6 @@ impl MsgId {
         Ok(result)
     }
 
-    /// Returns Some if the message needs to be moved from `folder`.
-    /// If yes, returns `ConfiguredInboxFolder`, `ConfiguredMvboxFolder` or `ConfiguredSentboxFolder`,
-    /// depending on where the message should be moved
-    pub async fn needs_move(self, context: &Context, folder: &str) -> Result<Option<Config>> {
-        use Config::*;
-        if context.is_mvbox(folder).await? {
-            return Ok(None);
-        }
-
-        let msg = Message::load_from_db(context, self).await?;
-
-        if context.is_spam_folder(folder).await? {
-            let msg_unblocked = msg.chat_id != DC_CHAT_ID_TRASH && msg.chat_blocked == Blocked::Not;
-
-            return if msg_unblocked {
-                if self.needs_move_to_mvbox(context, &msg).await? {
-                    Ok(Some(ConfiguredMvboxFolder))
-                } else {
-                    Ok(Some(ConfiguredInboxFolder))
-                }
-            } else {
-                // Blocked or contact request message in the spam folder, leave it there
-                Ok(None)
-            };
-        }
-
-        if self.needs_move_to_mvbox(context, &msg).await? {
-            Ok(Some(ConfiguredMvboxFolder))
-        } else if msg.state.is_outgoing()
-                && msg.is_dc_message == MessengerMessage::Yes
-                && !msg.is_setupmessage()
-                && msg.to_id != DC_CONTACT_ID_SELF // Leave self-chat-messages in the inbox, not sure about this
-                && context.is_inbox(folder).await?
-                && context.get_config_bool(SentboxMove).await?
-                && context.get_config(ConfiguredSentboxFolder).await?.is_some()
-        {
-            Ok(Some(ConfiguredSentboxFolder))
-        } else {
-            Ok(None)
-        }
-    }
-
-    async fn needs_move_to_mvbox(self, context: &Context, msg: &Message) -> Result<bool> {
-        if !context.get_config_bool(Config::MvboxMove).await? {
-            return Ok(false);
-        }
-
-        if msg.is_setupmessage() {
-            // do not move setup messages;
-            // there may be a non-delta device that wants to handle it
-            return Ok(false);
-        }
-
-        match msg.is_dc_message {
-            MessengerMessage::No => Ok(false),
-            MessengerMessage::Yes | MessengerMessage::Reply => Ok(true),
-        }
-    }
-
     /// Put message into trash chat and delete message text.
     ///
     /// It means the message is deleted locally, but not on the server.
@@ -282,6 +231,90 @@ impl Default for MessengerMessage {
     }
 }
 
+#[derive(Debug)]
+pub struct LazyMsg {
+    id: MsgId,
+    message: OnceCell<Message>,
+}
+
+impl LazyMsg {
+    // TODO should get() require &mut self? Or lock a mutex during computation?
+    async fn get(&self, context: &Context) -> Result<&Message> {
+        if let Some(m) = self.message.get() {
+            Ok(m)
+        } else {
+            let m = Message::load_from_db(context, self.id).await?;
+            self.message.set(m).ok();
+            self.message.get().context("message.set() failed???")
+        }
+    }
+
+    /// Returns Some if the message needs to be moved from `folder`.
+    /// If yes, returns `ConfiguredInboxFolder`, `ConfiguredMvboxFolder` or `ConfiguredSentboxFolder`,
+    /// depending on where the message should be moved
+    pub async fn needs_move(&self, context: &Context, folder: &str) -> Result<Option<Config>> {
+        use Config::*;
+        if context.disable_needs_move {
+            return Ok(None);
+        }
+
+        if context.is_mvbox(folder).await? {
+            return Ok(None);
+        }
+
+        let msg = self.get(context).await?;
+
+        if context.is_spam_folder(folder).await? {
+            let msg_unblocked = msg.chat_id != DC_CHAT_ID_TRASH && msg.chat_blocked == Blocked::Not;
+
+            return if msg_unblocked {
+                if self.needs_move_to_mvbox(context).await? {
+                    Ok(Some(ConfiguredMvboxFolder))
+                } else {
+                    Ok(Some(ConfiguredInboxFolder))
+                }
+            } else {
+                // Blocked or contact request message in the spam folder, leave it there
+                Ok(None)
+            };
+        }
+
+        if self.needs_move_to_mvbox(context).await? {
+            Ok(Some(ConfiguredMvboxFolder))
+        } else if msg.state.is_outgoing()
+                && msg.is_dc_message == MessengerMessage::Yes
+                && !msg.is_setupmessage()
+                && msg.to_id != DC_CONTACT_ID_SELF // Leave self-chat-messages in the inbox, not sure about this
+                && context.is_inbox(folder).await?
+                && context.get_config_bool(SentboxMove).await?
+                && context.get_config(ConfiguredSentboxFolder).await?.is_some()
+        {
+            Ok(Some(ConfiguredSentboxFolder))
+        } else {
+            Ok(None)
+        }
+    }
+
+    async fn needs_move_to_mvbox(&self, context: &Context) -> Result<bool> {
+        if !context.get_config_bool(Config::MvboxMove).await? {
+            return Ok(false);
+        }
+
+        let msg = self.get(context).await?;
+
+        if msg.is_setupmessage() {
+            // do not move setup messages;
+            // there may be a non-delta device that wants to handle it
+            return Ok(false);
+        }
+
+        match msg.is_dc_message {
+            MessengerMessage::No => Ok(false),
+            MessengerMessage::Yes | MessengerMessage::Reply => Ok(true),
+        }
+    }
+}
+
 /// An object representing a single message in memory.
 /// The message object is not updated.
 /// If you want an update, you have to recreate the object.
@@ -426,6 +459,13 @@ impl Message {
         Ok(msg)
     }
 
+    pub fn into_lazy(self) -> LazyMsg {
+        LazyMsg {
+            id: self.id,
+            message: self.into(),
+        }
+    }
+
     pub fn get_filemime(&self) -> Option<String> {
         if let Some(m) = self.param.get(Param::MimeType) {
             return Some(m.to_string());
@@ -1990,11 +2030,12 @@ mod tests {
         let exists = rfc724_mid_exists(&t, "abc@example.com").await.unwrap();
         let (folder_1, _, msg_id) = exists.unwrap();
         assert_eq!(folder, folder_1);
-        let actual = if let Some(config) = msg_id.needs_move(&t.ctx, folder).await.unwrap() {
-            t.ctx.get_config(config).await.unwrap()
-        } else {
-            None
-        };
+        let actual =
+            if let Some(config) = msg_id.to_lazy().needs_move(&t.ctx, folder).await.unwrap() {
+                t.ctx.get_config(config).await.unwrap()
+            } else {
+                None
+            };
         let expected = if expected_destination == folder {
             None
         } else {

src/sql.rs

@@ -3,7 +3,7 @@
 use async_std::path::Path;
 use async_std::sync::RwLock;
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::convert::TryFrom;
 use std::time::Duration;
 
@@ -39,12 +39,14 @@ mod migrations;
 #[derive(Debug)]
 pub struct Sql {
     pool: RwLock<Option<r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>>>,
+    config_cache: RwLock<HashMap<String, String>>, // TODO should be <String, Option<String>> in order to save that an option is None?
 }
 
 impl Default for Sql {
     fn default() -> Self {
         Self {
             pool: RwLock::new(None),
+            config_cache: RwLock::new(HashMap::new()),
         }
     }
 }
@@ -387,6 +389,8 @@ impl Sql {
     /// will already have been logged.
     pub async fn set_raw_config(&self, key: impl AsRef<str>, value: Option<&str>) -> Result<()> {
         let key = key.as_ref();
+
+        // TODO could be on another thread
         if let Some(value) = value {
             let exists = self
                 .exists(
@@ -413,20 +417,36 @@ impl Sql {
                 .await?;
         }
 
+        let mut lock = self.config_cache.write().await;
+        if let Some(v) = value {
+            lock.insert(key.to_string(), v.to_string());
+        } else {
+            lock.remove(key);
+        }
+        drop(lock);
+
         Ok(())
     }
 
     /// Get configuration options from the database.
     pub async fn get_raw_config(&self, key: impl AsRef<str>) -> Result<Option<String>> {
-        let value = self
-            .query_get_value(
-                "SELECT value FROM config WHERE keyname=?;",
-                paramsv![key.as_ref()],
-            )
-            .await
-            .context(format!("failed to fetch raw config: {}", key.as_ref()))?;
+        let lock = self.config_cache.read().await;
+        let cached = lock.get(key.as_ref()).cloned();
+        drop(lock);
 
-        Ok(value)
+        if let Some(c) = cached {
+            Ok(Some(c))
+        } else {
+            let value = self
+                .query_get_value(
+                    "SELECT value FROM config WHERE keyname=?;",
+                    paramsv![key.as_ref()],
+                )
+                .await
+                .context(format!("failed to fetch raw config: {}", key.as_ref()))?;
+
+            Ok(value)
+        }
     }
 
     pub async fn set_raw_config_int(&self, key: impl AsRef<str>, value: i32) -> Result<()> {

src/sql/migrations.rs

@@ -526,6 +526,10 @@ impl Sql {
         })
         .await?;
 
+        let mut lock = self.config_cache.write().await;
+        lock.insert(VERSION_CFG.to_string(), format!("{}", version));
+        drop(lock);
+
         Ok(())
     }
 }