Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-05-05 14:26:30 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix dc_job sql call, to reduce contention

Browse Source
This commit is contained in:
dignifiedquire
2019-07-11 10:58:58 +02:00
parent 3b27dd28b6
commit 3366eb147d
1 changed files with 115 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

227
src/dc_job.rs
View File

@@ -94,97 +94,103 @@ unsafe fn dc_job_perform(context: &Context, thread: libc::c_int, probe_network:
params_probe params_probe
}; };
context.sql.query_map(query, params, process_row, |jobs| { let jobs: Vec<dc_job_t> = context
for job in jobs { .sql
let mut job: dc_job_t = job?; .query_map(query, params, process_row, |jobs| {
jobs.collect::<Result<Vec<dc_job_t>, _>>()
.map_err(Into::into)
})
.unwrap_or_default();
for mut job in jobs {
info!(
context,
0,
"{}-job #{}, action {} started...",
if thread == 100 { "INBOX" } else { "SMTP" },
job.job_id,
job.action,
);
if 900 == job.action || 910 == job.action {
dc_job_kill_action(context, job.action);
dc_jobthread_suspend(context, &context.sentbox_thread.clone().read().unwrap(), 1);
dc_jobthread_suspend(context, &context.mvbox_thread.clone().read().unwrap(), 1);
dc_suspend_smtp_thread(context, 1);
}
let mut tries = 0;
while tries <= 1 {
job.try_again = 0;
match job.action {
5901 => {
dc_job_do_DC_JOB_SEND(context, &mut job);
}
110 => {
dc_job_do_DC_JOB_DELETE_MSG_ON_IMAP(context, &mut job);
}
130 => {
dc_job_do_DC_JOB_MARKSEEN_MSG_ON_IMAP(context, &mut job);
}
120 => {
dc_job_do_DC_JOB_MARKSEEN_MDN_ON_IMAP(context, &mut job);
}
200 => {
dc_job_do_DC_JOB_MOVE_MSG(context, &mut job);
}
5011 => {
dc_job_do_DC_JOB_SEND(context, &mut job);
}
900 => {
dc_job_do_DC_JOB_CONFIGURE_IMAP(context, &mut job);
}
910 => {
dc_job_do_DC_JOB_IMEX_IMAP(context, &mut job);
}
5005 => {
dc_job_do_DC_JOB_MAYBE_SEND_LOCATIONS(context, &mut job);
}
5007 => {
dc_job_do_DC_JOB_MAYBE_SEND_LOC_ENDED(context, &mut job);
}
105 => {
dc_housekeeping(context);
}
_ => {}
}
if job.try_again != -1 {
break;
}
tries += 1
}
if 900 == job.action || 910 == job.action {
dc_jobthread_suspend(
context,
&mut context.sentbox_thread.clone().read().unwrap(),
0,
);
dc_jobthread_suspend(
context,
&mut context.mvbox_thread.clone().read().unwrap(),
0,
);
dc_suspend_smtp_thread(context, 0);
break;
} else if job.try_again == 2 {
info!( info!(
context, context,
0, 0,
"{}-job #{}, action {} started...", "{}-job #{} not yet ready and will be delayed.",
if thread == 100 { "INBOX" } else { "SMTP" }, if thread == 100 { "INBOX" } else { "SMTP" },
job.job_id, job.job_id
job.action,
); );
} else if job.try_again == -1 || job.try_again == 3 {
if 900 == job.action || 910 == job.action { let tries = job.tries + 1;
dc_job_kill_action(context, job.action); if tries < 17 {
dc_jobthread_suspend(context, &context.sentbox_thread.clone().read().unwrap(), 1); job.tries = tries;
dc_jobthread_suspend(context, &context.mvbox_thread.clone().read().unwrap(), 1); let time_offset = get_backoff_time_offset(tries);
dc_suspend_smtp_thread(context, 1); job.desired_timestamp = job.added_timestamp + time_offset;
} dc_job_update(context, &mut job);
let mut tries = 0;
while tries <= 1 {
job.try_again = 0;
match job.action {
5901 => {
dc_job_do_DC_JOB_SEND(context, &mut job);
}
110 => {
dc_job_do_DC_JOB_DELETE_MSG_ON_IMAP(context, &mut job);
}
130 => {
dc_job_do_DC_JOB_MARKSEEN_MSG_ON_IMAP(context, &mut job);
}
120 => {
dc_job_do_DC_JOB_MARKSEEN_MDN_ON_IMAP(context, &mut job);
}
200 => {
dc_job_do_DC_JOB_MOVE_MSG(context, &mut job);
}
5011 => {
dc_job_do_DC_JOB_SEND(context, &mut job);
}
900 => {
dc_job_do_DC_JOB_CONFIGURE_IMAP(context, &mut job);
}
910 => {
dc_job_do_DC_JOB_IMEX_IMAP(context, &mut job);
}
5005 => {
dc_job_do_DC_JOB_MAYBE_SEND_LOCATIONS(context, &mut job);
}
5007 => {
dc_job_do_DC_JOB_MAYBE_SEND_LOC_ENDED(context, &mut job);
}
105 => {
dc_housekeeping(context);
}
_ => {}
}
if job.try_again != -1 {
break;
}
tries += 1
}
if 900 == job.action || 910 == job.action {
dc_jobthread_suspend(
context,
&mut context.sentbox_thread.clone().read().unwrap(),
0,
);
dc_jobthread_suspend(
context,
&mut context.mvbox_thread.clone().read().unwrap(),
0,
);
dc_suspend_smtp_thread(context, 0);
break;
} else if job.try_again == 2 {
info!( info!(
context,
0,
"{}-job #{} not yet ready and will be delayed.",
if thread == 100 { "INBOX" } else { "SMTP" },
job.job_id
);
} else if job.try_again == -1 || job.try_again == 3 {
let tries = job.tries + 1;
if tries < 17 {
job.tries = tries;
let time_offset = get_backoff_time_offset(tries);
job.desired_timestamp = job.added_timestamp + time_offset;
dc_job_update(context, &mut job);
info!(
context, context,
0, 0,
"{}-job #{} not succeeded on try #{}, retry in ADD_TIME+{} (in {} seconds).", "{}-job #{} not succeeded on try #{}, retry in ADD_TIME+{} (in {} seconds).",
@@ -194,38 +200,35 @@ unsafe fn dc_job_perform(context: &Context, thread: libc::c_int, probe_network:
time_offset, time_offset,
job.added_timestamp + time_offset - time() job.added_timestamp + time_offset - time()
); );
if thread == 5000 && tries < 17 - 1 { if thread == 5000 && tries < 17 - 1 {
context context
.smtp_state .smtp_state
.clone() .clone()
.0 .0
.lock() .lock()
.unwrap() .unwrap()
.perform_jobs_needed = 2; .perform_jobs_needed = 2;
}
} else {
if job.action == 5901 {
dc_set_msg_failed(context, job.foreign_id, job.pending_error);
}
dc_job_delete(context, &mut job);
} }
if 0 == probe_network {
continue;
}
// on dc_maybe_network() we stop trying here;
// these jobs are already tried once.
// otherwise, we just continue with the next job
// to give other jobs a chance being tried at least once.
break;
} else { } else {
if job.action == 5901 {
dc_set_msg_failed(context, job.foreign_id, job.pending_error);
}
dc_job_delete(context, &mut job); dc_job_delete(context, &mut job);
} }
dc_param_unref(job.param); if 0 == probe_network {
free(job.pending_error as *mut libc::c_void); continue;
}
// on dc_maybe_network() we stop trying here;
// these jobs are already tried once.
// otherwise, we just continue with the next job
// to give other jobs a chance being tried at least once.
break;
} else {
dc_job_delete(context, &mut job);
} }
dc_param_unref(job.param);
Ok(()) free(job.pending_error as *mut libc::c_void);
}); // TODO: better error handling }
} }
fn dc_job_delete(context: &Context, job: &dc_job_t) -> bool { fn dc_job_delete(context: &Context, job: &dc_job_t) -> bool {