From: Timo Sirainen Date: Sun, 30 Apr 2017 09:31:48 +0000 (+0300) Subject: dsync: Try to commit transactions every dsync_commit_msgs_interval messages X-Git-Tag: 2.3.0.rc1~1622 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=a76faea3eb26c4cd67886fbe02c604f74d54be8c;p=thirdparty%2Fdovecot%2Fcore.git dsync: Try to commit transactions every dsync_commit_msgs_interval messages This was first attempted to be implemented by ec0cc8fa647794e44a1afaa448f495a713048dc4, but it was later partially reverted by 5973d496b16721af6d2c1fa90b016aacddf13554. This current commit should fix its problems. --- diff --git a/src/doveadm/doveadm-dsync.c b/src/doveadm/doveadm-dsync.c index 75f1229788..caf569f1d5 100644 --- a/src/doveadm/doveadm-dsync.c +++ b/src/doveadm/doveadm-dsync.c @@ -93,6 +93,7 @@ struct dsync_cmd_context { const char *error; unsigned int lock_timeout; + unsigned int import_commit_msgs_interval; bool lock:1; bool purge_remote:1; @@ -592,6 +593,7 @@ cmd_dsync_run(struct doveadm_mail_cmd_context *_ctx, struct mail_user *user) set.virtual_all_box = ctx->virtual_all_box; memcpy(set.sync_box_guid, ctx->mailbox_guid, sizeof(set.sync_box_guid)); set.lock_timeout_secs = ctx->lock_timeout; + set.import_commit_msgs_interval = ctx->import_commit_msgs_interval; set.state = ctx->state_input; set.mailbox_alt_char = doveadm_settings->dsync_alt_char[0]; @@ -1111,6 +1113,7 @@ static struct doveadm_mail_cmd_context *cmd_dsync_alloc(void) p_array_init(&ctx->namespace_prefixes, ctx->ctx.pool, 4); if ((doveadm_settings->parsed_features & DSYNC_FEATURE_EMPTY_HDR_WORKAROUND) != 0) ctx->empty_hdr_workaround = TRUE; + ctx->import_commit_msgs_interval = doveadm_settings->dsync_commit_msgs_interval; return &ctx->ctx; } diff --git a/src/doveadm/doveadm-settings.c b/src/doveadm/doveadm-settings.c index c41d013c0f..9c031b7288 100644 --- a/src/doveadm/doveadm-settings.c +++ b/src/doveadm/doveadm-settings.c @@ -72,6 +72,7 @@ static const struct setting_define doveadm_setting_defines[] = { DEF(SET_STR, director_username_hash), DEF(SET_STR, doveadm_api_key), DEF(SET_STR, dsync_features), + DEF(SET_UINT, dsync_commit_msgs_interval), DEF(SET_STR, doveadm_http_rawlog_dir), { SET_STRLIST, "plugin", offsetof(struct doveadm_settings, plugin_envs), NULL }, @@ -95,6 +96,7 @@ const struct doveadm_settings doveadm_default_settings = { .dsync_alt_char = "_", .dsync_remote_cmd = "ssh -l%{login} %{host} doveadm dsync-server -u%u -U", .dsync_features = "", + .dsync_commit_msgs_interval = 100, .ssl_client_ca_dir = "", .ssl_client_ca_file = "", .director_username_hash = "%Lu", diff --git a/src/doveadm/doveadm-settings.h b/src/doveadm/doveadm-settings.h index 8bfa39c702..2990a91e8f 100644 --- a/src/doveadm/doveadm-settings.h +++ b/src/doveadm/doveadm-settings.h @@ -29,6 +29,7 @@ struct doveadm_settings { const char *director_username_hash; const char *doveadm_api_key; const char *dsync_features; + unsigned int dsync_commit_msgs_interval; const char *doveadm_http_rawlog_dir; enum dsync_features parsed_features; ARRAY(const char *) plugin_envs; diff --git a/src/doveadm/dsync/dsync-brain-mailbox.c b/src/doveadm/dsync/dsync-brain-mailbox.c index ab76023904..937338347e 100644 --- a/src/doveadm/dsync/dsync-brain-mailbox.c +++ b/src/doveadm/dsync/dsync-brain-mailbox.c @@ -236,6 +236,7 @@ dsync_brain_sync_mailbox_init_remote(struct dsync_brain *brain, brain->sync_until_timestamp, brain->sync_max_size, brain->sync_flag, + brain->import_commit_msgs_interval, import_flags); } diff --git a/src/doveadm/dsync/dsync-brain-private.h b/src/doveadm/dsync/dsync-brain-private.h index 1de0313baa..b97b578364 100644 --- a/src/doveadm/dsync/dsync-brain-private.h +++ b/src/doveadm/dsync/dsync-brain-private.h @@ -62,6 +62,7 @@ struct dsync_brain { uoff_t sync_max_size; const char *sync_flag; char alt_char; + unsigned int import_commit_msgs_interval; unsigned int lock_timeout; int lock_fd; diff --git a/src/doveadm/dsync/dsync-brain.c b/src/doveadm/dsync/dsync-brain.c index 17f695a912..cb8d290f92 100644 --- a/src/doveadm/dsync/dsync-brain.c +++ b/src/doveadm/dsync/dsync-brain.c @@ -221,6 +221,7 @@ dsync_brain_master_init(struct mail_user *user, struct dsync_ibc *ibc, memcpy(brain->sync_box_guid, set->sync_box_guid, sizeof(brain->sync_box_guid)); brain->lock_timeout = set->lock_timeout_secs; + brain->import_commit_msgs_interval = set->import_commit_msgs_interval; brain->master_brain = TRUE; dsync_brain_set_flags(brain, flags); @@ -260,6 +261,7 @@ dsync_brain_master_init(struct mail_user *user, struct dsync_ibc *ibc, ibc_set.sync_type = sync_type; ibc_set.hdr_hash_v2 = TRUE; ibc_set.lock_timeout = set->lock_timeout_secs; + ibc_set.import_commit_msgs_interval = set->import_commit_msgs_interval; /* reverse the backup direction for the slave */ ibc_set.brain_flags = flags & ~(DSYNC_BRAIN_FLAG_BACKUP_SEND | DSYNC_BRAIN_FLAG_BACKUP_RECV); diff --git a/src/doveadm/dsync/dsync-brain.h b/src/doveadm/dsync/dsync-brain.h index 5027378a11..929b80344b 100644 --- a/src/doveadm/dsync/dsync-brain.h +++ b/src/doveadm/dsync/dsync-brain.h @@ -79,6 +79,9 @@ struct dsync_brain_settings { /* If non-zero, use dsync lock file for this user */ unsigned int lock_timeout_secs; + /* If non-zero, importing will attempt to commit transaction after + saving this many messages. */ + unsigned int import_commit_msgs_interval; /* Input state for DSYNC_BRAIN_SYNC_TYPE_STATE */ const char *state; }; diff --git a/src/doveadm/dsync/dsync-ibc-stream.c b/src/doveadm/dsync/dsync-ibc-stream.c index ecd6327cd2..eebde83d6d 100644 --- a/src/doveadm/dsync/dsync-ibc-stream.c +++ b/src/doveadm/dsync/dsync-ibc-stream.c @@ -78,7 +78,7 @@ static const struct { "send_mail_requests backup_send backup_recv lock_timeout " "no_mail_sync no_mailbox_renames no_backup_overwrite purge_remote " "no_notify sync_since_timestamp sync_max_size sync_flags sync_until_timestamp" - "virtual_all_box empty_hdr_workaround" + "virtual_all_box empty_hdr_workaround import_commit_msgs_interval" }, { .name = "mailbox_state", .chr = 'S', @@ -704,6 +704,10 @@ dsync_ibc_stream_send_handshake(struct dsync_ibc *_ibc, dsync_serializer_encode_add(encoder, "lock_timeout", t_strdup_printf("%u", set->lock_timeout)); } + if (set->import_commit_msgs_interval > 0) { + dsync_serializer_encode_add(encoder, "import_commit_msgs_interval", + t_strdup_printf("%u", set->import_commit_msgs_interval)); + } if (set->sync_since_timestamp > 0) { dsync_serializer_encode_add(encoder, "sync_since_timestamp", t_strdup_printf("%ld", (long)set->sync_since_timestamp)); @@ -820,6 +824,14 @@ dsync_ibc_stream_recv_handshake(struct dsync_ibc *_ibc, return DSYNC_IBC_RECV_RET_TRYAGAIN; } } + if (dsync_deserializer_decode_try(decoder, "import_commit_msgs_interval", &value)) { + if (str_to_uint(value, &set->import_commit_msgs_interval) < 0 || + set->import_commit_msgs_interval == 0) { + dsync_ibc_input_error(ibc, decoder, + "Invalid import_commit_msgs_interval: %s", value); + return DSYNC_IBC_RECV_RET_TRYAGAIN; + } + } if (dsync_deserializer_decode_try(decoder, "sync_since_timestamp", &value)) { if (str_to_time(value, &set->sync_since_timestamp) < 0 || set->sync_since_timestamp == 0) { diff --git a/src/doveadm/dsync/dsync-ibc.h b/src/doveadm/dsync/dsync-ibc.h index db02a04aae..7cc8ece9d8 100644 --- a/src/doveadm/dsync/dsync-ibc.h +++ b/src/doveadm/dsync/dsync-ibc.h @@ -69,6 +69,7 @@ struct dsync_ibc_settings { enum dsync_brain_flags brain_flags; bool hdr_hash_v2; unsigned int lock_timeout; + unsigned int import_commit_msgs_interval; }; void dsync_ibc_init_pipe(struct dsync_ibc **ibc1_r, diff --git a/src/doveadm/dsync/dsync-mailbox-import.c b/src/doveadm/dsync/dsync-mailbox-import.c index 16ddab8353..f3f7032ae1 100644 --- a/src/doveadm/dsync/dsync-mailbox-import.c +++ b/src/doveadm/dsync/dsync-mailbox-import.c @@ -44,6 +44,7 @@ struct importer_new_mail { bool skip:1; bool expunged:1; bool copy_failed:1; + bool saved:1; }; /* for quickly testing that two-way sync doesn't actually do any unexpected @@ -66,6 +67,7 @@ struct dsync_mailbox_importer { uoff_t sync_max_size; enum mailbox_transaction_flags transaction_flags; unsigned int hdr_hash_version; + unsigned int commit_msgs_interval; enum mail_flags sync_flag; const char *sync_keyword; @@ -106,6 +108,7 @@ struct dsync_mailbox_importer { uint32_t prev_uid, next_local_seq, local_uid_next; uint64_t local_initial_highestmodseq, local_initial_highestpvtmodseq; unsigned int import_pos, import_count; + unsigned int first_unsaved_idx, saves_since_commit; enum mail_error mail_error; @@ -222,6 +225,7 @@ dsync_mailbox_import_init(struct mailbox *box, time_t sync_until_timestamp, uoff_t sync_max_size, const char *sync_flag, + unsigned int commit_msgs_interval, enum dsync_mailbox_import_flags flags) { struct dsync_mailbox_importer *importer; @@ -257,6 +261,7 @@ dsync_mailbox_import_init(struct mailbox *box, else importer->sync_keyword = p_strdup(pool, sync_flag); } + importer->commit_msgs_interval = commit_msgs_interval; importer->transaction_flags = MAILBOX_TRANSACTION_FLAG_SYNC; if ((flags & DSYNC_MAILBOX_IMPORT_FLAG_NO_NOTIFY) != 0) importer->transaction_flags |= MAILBOX_TRANSACTION_FLAG_NO_NOTIFY; @@ -1817,6 +1822,17 @@ int dsync_mailbox_import_change(struct dsync_mailbox_importer *importer, return importer->failed ? -1 : 0; } +static int +importer_new_mail_final_uid_cmp(struct importer_new_mail *const *newmail1, + struct importer_new_mail *const *newmail2) +{ + if ((*newmail1)->final_uid < (*newmail2)->final_uid) + return -1; + if ((*newmail1)->final_uid > (*newmail2)->final_uid) + return 1; + return 0; +} + static void dsync_mailbox_import_assign_new_uids(struct dsync_mailbox_importer *importer) { @@ -1829,6 +1845,7 @@ dsync_mailbox_import_assign_new_uids(struct dsync_mailbox_importer *importer) newmail = *newmailp; if (newmail->skip) { /* already assigned */ + i_assert(newmail->final_uid != 0); continue; } @@ -1856,6 +1873,9 @@ dsync_mailbox_import_assign_new_uids(struct dsync_mailbox_importer *importer) } importer->last_common_uid = common_uid_next-1; importer->new_uids_assigned = TRUE; + /* Sort the newmails by their final_uid. This is used for tracking + whether an intermediate commit is allowed. */ + array_sort(&importer->newmails, importer_new_mail_final_uid_cmp); } static int @@ -1901,6 +1921,45 @@ dsync_mailbox_import_saved_uid(struct dsync_mailbox_importer *importer, array_append(&importer->wanted_uids, &uid, 1); } +static void +dsync_mailbox_import_update_first_saved(struct dsync_mailbox_importer *importer) +{ + struct importer_new_mail *const *newmails; + unsigned int count; + + newmails = array_get(&importer->newmails, &count); + while (importer->first_unsaved_idx < count) { + if (!newmails[importer->first_unsaved_idx]->saved) + break; + importer->first_unsaved_idx++; + } +} + +static void +dsync_mailbox_import_saved_newmail(struct dsync_mailbox_importer *importer, + struct importer_new_mail *newmail) +{ + dsync_mailbox_import_saved_uid(importer, newmail->final_uid); + newmail->saved = TRUE; + + dsync_mailbox_import_update_first_saved(importer); + importer->saves_since_commit++; + /* we can commit only if all the upcoming mails will have UIDs that + are larger than we're committing. + + Note that if any existing UIDs have been changed, the new UID is + usually higher than anything that is being saved so we can't do + an intermediate commit. It's too much extra work to try to handle + that situation. So here this never happens, because then + array_count(wanted_uids) is always higher than first_unsaved_idx. */ + if (importer->saves_since_commit >= importer->commit_msgs_interval && + importer->first_unsaved_idx == array_count(&importer->wanted_uids)) { + if (dsync_mailbox_import_commit(importer, FALSE) < 0) + importer->failed = TRUE; + importer->saves_since_commit = 0; + } +} + static bool dsync_msg_change_uid(struct dsync_mailbox_importer *importer, uint32_t old_uid, uint32_t new_uid) @@ -2367,7 +2426,7 @@ dsync_mailbox_save_body(struct dsync_mailbox_importer *importer, } if (ret > 0) { i_assert(save_ctx == NULL); - dsync_mailbox_import_saved_uid(importer, newmail->final_uid); + dsync_mailbox_import_saved_newmail(importer, newmail); return TRUE; } /* fallback to saving from remote stream */ @@ -2447,8 +2506,7 @@ dsync_mailbox_save_body(struct dsync_mailbox_importer *importer, &importer->mail_error)); importer->failed = TRUE; } else { - dsync_mailbox_import_saved_uid(importer, - newmail->final_uid); + dsync_mailbox_import_saved_newmail(importer, newmail); } } return TRUE; diff --git a/src/doveadm/dsync/dsync-mailbox-import.h b/src/doveadm/dsync/dsync-mailbox-import.h index 50d74abf6c..b9babe5c11 100644 --- a/src/doveadm/dsync/dsync-mailbox-import.h +++ b/src/doveadm/dsync/dsync-mailbox-import.h @@ -36,6 +36,7 @@ dsync_mailbox_import_init(struct mailbox *box, time_t sync_until_timestamp, uoff_t sync_max_size, const char *sync_flag, + unsigned int commit_msgs_interval, enum dsync_mailbox_import_flags flags); int dsync_mailbox_import_attribute(struct dsync_mailbox_importer *importer, const struct dsync_mailbox_attribute *attr);