imap: count message totals using HashSet

This way it's easy to know if a flag change in an envelope requires the
unseen total of a mailbox to change.
memfd
Manos Pitsidianakis 2020-06-24 17:47:08 +03:00
parent c4bc7be5d1
commit 91badc3960
Signed by: Manos Pitsidianakis
GPG Key ID: 73627C2F690DF710
4 changed files with 133 additions and 42 deletions

View File

@ -46,7 +46,7 @@ use crate::conf::AccountSettings;
use crate::email::*;
use crate::error::{MeliError, Result, ResultIntoMeliError};
use std::collections::{hash_map::DefaultHasher, BTreeMap};
use std::collections::{HashMap, HashSet};
use std::collections::{BTreeSet, HashMap, HashSet};
use std::hash::Hasher;
use std::str::FromStr;
use std::sync::{Arc, Mutex, RwLock};
@ -236,7 +236,7 @@ impl MailBackend for ImapType {
let _tx = tx.clone();
if let Err(err) = (move || {
let tx = _tx;
let mut our_unseen = 0;
let mut our_unseen: BTreeSet<EnvelopeHash> = Default::default();
let mut valid_hash_set: HashSet<EnvelopeHash> = HashSet::default();
let cached_hash_set: HashSet<EnvelopeHash> =
(|| -> Result<HashSet<EnvelopeHash>> {
@ -266,8 +266,8 @@ impl MailBackend for ImapType {
if !envelopes.is_empty() {
let mut payload = vec![];
for (uid, env) in envelopes {
if !env.flags().contains(Flag::SEEN) {
our_unseen += 1;
if !env.is_seen() {
our_unseen.insert(env.hash());
}
uid_store
.hash_index
@ -283,7 +283,7 @@ impl MailBackend for ImapType {
}
debug!("sending cached payload for {}", mailbox_hash);
*unseen.lock().unwrap() = our_unseen;
unseen.lock().unwrap().insert_set(our_unseen.clone());
tx.send(AsyncStatus::Payload(Ok(payload))).unwrap();
}
Ok(ret)
@ -334,8 +334,10 @@ impl MailBackend for ImapType {
permissions.rename_messages = !examine_response.read_only;
permissions.delete_messages = !examine_response.read_only;
permissions.delete_messages = !examine_response.read_only;
let mut mailbox_exists = mailbox_exists.lock().unwrap();
*mailbox_exists = examine_response.exists;
mailbox_exists
.lock()
.unwrap()
.set_not_yet_seen(examine_response.exists);
}
if examine_response.exists == 0 {
if uid_store.cache_headers {
@ -429,8 +431,8 @@ impl MailBackend for ImapType {
env.set_hash(h.finish());
valid_hash_set.insert(env.hash());
if let Some((flags, keywords)) = flags {
if !flags.contains(Flag::SEEN) {
our_unseen += 1;
if !flags.intersects(Flag::SEEN) {
our_unseen.insert(env.hash());
}
env.set_flags(flags);
for f in keywords {
@ -487,8 +489,14 @@ impl MailBackend for ImapType {
kind: RefreshEventKind::Remove(env_hash),
});
}
*unseen.lock().unwrap() = our_unseen;
let progress = envelopes.len();
unseen
.lock()
.unwrap()
.insert_set(our_unseen.iter().cloned().collect());
mailbox_exists.lock().unwrap().insert_existing_set(
envelopes.iter().map(|(_, env)| env.hash()).collect::<_>(),
);
tx.send(AsyncStatus::Payload(Ok(envelopes
.into_iter()
.map(|(_, env)| env)

View File

@ -21,9 +21,78 @@
use crate::backends::{
BackendMailbox, Mailbox, MailboxHash, MailboxPermissions, SpecialUsageMailbox,
};
use crate::email::EnvelopeHash;
use crate::error::*;
use std::collections::BTreeSet;
use std::sync::{Arc, Mutex, RwLock};
#[derive(Debug, Default, Clone)]
pub struct LazyCountSet {
not_yet_seen: usize,
set: BTreeSet<EnvelopeHash>,
}
impl LazyCountSet {
pub fn set_not_yet_seen(&mut self, new_val: usize) {
self.not_yet_seen = new_val;
}
pub fn insert_existing(&mut self, new_val: EnvelopeHash) -> bool {
if self.not_yet_seen == 0 {
false
} else {
self.not_yet_seen -= 1;
self.set.insert(new_val);
true
}
}
pub fn insert_existing_set(&mut self, set: BTreeSet<EnvelopeHash>) -> bool {
debug!("insert_existing_set {:?}", &set);
if self.not_yet_seen < set.len() {
false
} else {
self.not_yet_seen -= set.len();
self.set.extend(set.into_iter());
true
}
}
#[inline(always)]
pub fn len(&self) -> usize {
self.set.len() + self.not_yet_seen
}
#[inline(always)]
pub fn clear(&mut self) {
self.set.clear();
self.not_yet_seen = 0;
}
pub fn insert_new(&mut self, new_val: EnvelopeHash) {
self.set.insert(new_val);
}
pub fn insert_set(&mut self, set: BTreeSet<EnvelopeHash>) {
debug!("insert__set {:?}", &set);
self.set.extend(set.into_iter());
}
pub fn remove(&mut self, new_val: EnvelopeHash) -> bool {
self.set.remove(&new_val)
}
}
#[test]
fn test_lazy_count_set() {
let mut new = LazyCountSet::default();
new.set_not_yet_seen(10);
for i in 0..10 {
assert!(new.insert_existing(i));
}
assert!(!new.insert_existing(10));
}
#[derive(Debug, Default, Clone)]
pub struct ImapMailbox {
pub(super) hash: MailboxHash,
@ -38,8 +107,8 @@ pub struct ImapMailbox {
pub is_subscribed: bool,
pub permissions: Arc<Mutex<MailboxPermissions>>,
pub exists: Arc<Mutex<usize>>,
pub unseen: Arc<Mutex<usize>>,
pub exists: Arc<Mutex<LazyCountSet>>,
pub unseen: Arc<Mutex<LazyCountSet>>,
}
impl ImapMailbox {
@ -98,6 +167,6 @@ impl BackendMailbox for ImapMailbox {
}
fn count(&self) -> Result<(usize, usize)> {
Ok((*self.unseen.lock()?, *self.exists.lock()?))
Ok((self.unseen.lock()?.len(), self.exists.lock()?.len()))
}
}

View File

@ -106,13 +106,13 @@ impl ImapConnection {
* */
let mut prev_exists = mailbox.exists.lock().unwrap();
debug!("exists {}", n);
if n > *prev_exists {
if n > prev_exists.len() {
try_fail!(
mailbox_hash,
self.send_command(
&[
b"FETCH",
format!("{}:{}", *prev_exists + 1, n).as_bytes(),
format!("{}:{}", prev_exists.len() + 1, n).as_bytes(),
b"(UID FLAGS RFC822)",
]
.join(&b' '),
@ -165,8 +165,9 @@ impl ImapConnection {
mailbox.path(),
);
if !env.is_seen() {
*mailbox.unseen.lock().unwrap() += 1;
mailbox.unseen.lock().unwrap().insert_new(env.hash());
}
prev_exists.insert_new(env.hash());
self.add_refresh_event(RefreshEvent {
account_hash: self.uid_store.account_hash,
mailbox_hash,
@ -179,9 +180,6 @@ impl ImapConnection {
debug!(e);
}
}
*prev_exists = n;
} else if n < *prev_exists {
*prev_exists = n;
}
}
UntaggedResponse::Recent(_) => {
@ -213,7 +211,6 @@ impl ImapConnection {
uid, flags, body, ..
} in v
{
*mailbox.exists.lock().unwrap() += 1;
if !self
.uid_store
.uid_index
@ -253,9 +250,14 @@ impl ImapConnection {
}
}
if !env.is_seen() {
*mailbox.unseen.lock().unwrap() += 1;
mailbox
.unseen
.lock()
.unwrap()
.insert_new(env.hash());
}
mailbox.exists.lock().unwrap().insert_new(env.hash());
self.add_refresh_event(RefreshEvent {
account_hash: self.uid_store.account_hash,
mailbox_hash,
@ -307,6 +309,11 @@ impl ImapConnection {
let env_hash = lck.get(&(mailbox_hash, uid)).map(|&h| h);
drop(lck);
if let Some(env_hash) = env_hash {
if !flags.0.intersects(crate::email::Flag::SEEN) {
mailbox.unseen.lock().unwrap().insert_new(env_hash);
} else {
mailbox.unseen.lock().unwrap().remove(env_hash);
}
self.add_refresh_event(RefreshEvent {
account_hash: self.uid_store.account_hash,
mailbox_hash,

View File

@ -155,7 +155,7 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
debug!("select response {}", &response);
{
let mut prev_exists = mailbox.exists.lock().unwrap();
*prev_exists = match protocol_parser::select_response(&response) {
match protocol_parser::select_response(&response) {
Ok(ok) => {
{
uidvalidity = ok.uidvalidity;
@ -168,7 +168,7 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
mailbox_hash,
kind: RefreshEventKind::Rescan,
});
*prev_exists = 0;
prev_exists.clear();
/*
uid_store.uid_index.lock().unwrap().clear();
uid_store.hash_index.lock().unwrap().clear();
@ -194,7 +194,6 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
}
}
debug!(&ok);
ok.exists
}
Err(e) => {
debug!("{:?}", e);
@ -322,7 +321,6 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
))
.unwrap();
ctr += 1;
*mailbox.exists.lock().unwrap() += 1;
if !uid_store
.uid_index
.lock()
@ -361,7 +359,11 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
}
}
if !env.is_seen() {
*mailbox.unseen.lock().unwrap() += 1;
mailbox
.unseen
.lock()
.unwrap()
.insert_new(env.hash());
}
if uid_store.cache_headers {
cache::save_envelopes(
@ -371,6 +373,7 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
&[(uid, &env)],
)?;
}
mailbox.exists.lock().unwrap().insert_new(env.hash());
conn.add_refresh_event(RefreshEvent {
account_hash: uid_store.account_hash,
@ -445,12 +448,12 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
format!(
"got `{} EXISTS` notification (EXISTS was previously {} for {}",
n,
*prev_exists,
prev_exists.len(),
mailbox.path()
),
))
.unwrap();
if n > *prev_exists {
if n > prev_exists.len() {
exit_on_error!(
conn,
mailbox_hash,
@ -460,7 +463,7 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
conn.send_command(
&[
b"FETCH",
format!("{}:{}", *prev_exists + 1, n).as_bytes(),
format!("{}:{}", prev_exists.len() + 1, n).as_bytes(),
b"(UID FLAGS RFC822)",
]
.join(&b' '),
@ -523,7 +526,7 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
mailbox.path(),
);
if !env.is_seen() {
*mailbox.unseen.lock().unwrap() += 1;
mailbox.unseen.lock().unwrap().insert_new(env.hash());
}
if uid_store.cache_headers {
cache::save_envelopes(
@ -533,6 +536,7 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
&[(uid, &env)],
)?;
}
prev_exists.insert_new(env.hash());
conn.add_refresh_event(RefreshEvent {
account_hash: uid_store.account_hash,
@ -550,10 +554,6 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
debug!(e);
}
}
*prev_exists = n;
} else if n < *prev_exists {
*prev_exists = n;
}
}
Ok(Some(Fetch(msg_seq, flags))) => {
@ -588,6 +588,11 @@ pub fn idle(kit: ImapWatchKit) -> Result<()> {
.unwrap()
.get(&(mailbox_hash, uid))
{
if !flags.0.intersects(crate::email::Flag::SEEN) {
mailbox.unseen.lock().unwrap().insert_new(*env_hash);
} else {
mailbox.unseen.lock().unwrap().remove(*env_hash);
}
conn.add_refresh_event(RefreshEvent {
account_hash: uid_store.account_hash,
mailbox_hash,
@ -767,7 +772,11 @@ pub fn examine_updates(
}
}
if !env.is_seen() {
*mailbox.unseen.lock().unwrap() += 1;
mailbox
.unseen
.lock()
.unwrap()
.insert_new(env.hash());
}
if uid_store.cache_headers {
cache::save_envelopes(
@ -777,6 +786,7 @@ pub fn examine_updates(
&[(uid, &env)],
)?;
}
prev_exists.insert_new(env.hash());
conn.add_refresh_event(RefreshEvent {
account_hash: uid_store.account_hash,
@ -800,7 +810,7 @@ pub fn examine_updates(
}
}
}
} else if n > *prev_exists {
} else if n > prev_exists.len() {
/* UID FETCH ALL UID, cross-ref, then FETCH difference headers
* */
debug!("exists {}", n);
@ -812,7 +822,7 @@ pub fn examine_updates(
conn.send_command(
&[
b"FETCH",
format!("{}:{}", *prev_exists + 1, n).as_bytes(),
format!("{}:{}", prev_exists.len() + 1, n).as_bytes(),
b"(UID FLAGS RFC822)",
]
.join(&b' '),
@ -863,7 +873,7 @@ pub fn examine_updates(
mailbox.path(),
);
if !env.is_seen() {
*mailbox.unseen.lock().unwrap() += 1;
mailbox.unseen.lock().unwrap().insert_new(env.hash());
}
if uid_store.cache_headers {
cache::save_envelopes(
@ -873,6 +883,7 @@ pub fn examine_updates(
&[(uid, &env)],
)?;
}
prev_exists.insert_new(env.hash());
conn.add_refresh_event(RefreshEvent {
account_hash: uid_store.account_hash,
@ -886,10 +897,6 @@ pub fn examine_updates(
debug!(e);
}
}
*prev_exists = n;
} else if n < *prev_exists {
*prev_exists = n;
}
}
Err(e) => {