replicator: doveadm commands and user list export may have skipped some users.

The users were exported from the queue, but they are temporarily removed
from there while the user is being replicated. The users always exist in the
hash table though.

diff --git a/src/replication/replicator/doveadm-connection.c b/src/replication/replicator/doveadm-connection.c
index 3b16f1202bb93405ae1ddf6ffd1c5dd96d4b3953..46c44e17e64af9e5a47400bf693b07bae4708f29 100644 (file)
--- a/src/replication/replicator/doveadm-connection.c
+++ b/src/replication/replicator/doveadm-connection.c
@@ -23,10 +23,11 @@ static struct connection_list *doveadm_connections;
 
 static int client_input_status_overview(struct doveadm_connection *client)
 {
-       struct replicator_user *const *users;
+       struct replicator_queue_iter *iter;
+       struct replicator_user *user;
        enum replication_priority priority;
        unsigned int pending_counts[REPLICATION_PRIORITY_SYNC+1];
-       unsigned int i, count, next_secs, pending_failed_count;
+       unsigned int user_count, next_secs, pending_failed_count;
        unsigned int pending_full_resync_count, waiting_failed_count;
        string_t *str = t_str_new(256);
 
@@ -34,21 +35,24 @@ static int client_input_status_overview(struct doveadm_connection *client)
        pending_failed_count = 0; waiting_failed_count = 0;
        pending_full_resync_count = 0;
 
-       users = replicator_queue_get_users(client->queue, &count);
-       for (i = 0; i < count; i++) {
-               if (users[i]->priority != REPLICATION_PRIORITY_NONE)
-                       pending_counts[users[i]->priority]++;
+       user_count = 0;
+       iter = replicator_queue_iter_init(client->queue);
+       while ((user = replicator_queue_iter_next(iter)) != NULL) {
+               if (user->priority != REPLICATION_PRIORITY_NONE)
+                       pending_counts[user->priority]++;
                else if (replicator_queue_want_sync_now(client->queue,
-                                                       users[i], &next_secs)) {
-                       if (users[i]->last_sync_failed)
+                                                       user, &next_secs)) {
+                       if (user->last_sync_failed)
                                pending_failed_count++;
                        else
                                pending_full_resync_count++;
                } else {
-                       if (users[i]->last_sync_failed)
+                       if (user->last_sync_failed)
                                waiting_failed_count++;
                }
+               user_count++;
        }
+       replicator_queue_iter_deinit(&iter);
 
        for (priority = REPLICATION_PRIORITY_SYNC; priority > 0; priority--) {
                str_printfa(str, "Queued '%s' requests\t%u\n",
@@ -61,7 +65,7 @@ static int client_input_status_overview(struct doveadm_connection *client)
                    pending_full_resync_count);
        str_printfa(str, "Waiting 'failed' requests\t%u\n",
                    waiting_failed_count);
-       str_printfa(str, "Total number of known users\t%u\n", count);
+       str_printfa(str, "Total number of known users\t%u\n", user_count);
        str_append_c(str, '\n');
        o_stream_send(client->conn.output, str_data(str), str_len(str));
        return 0;
@@ -70,17 +74,16 @@ static int client_input_status_overview(struct doveadm_connection *client)
 static int
 client_input_status(struct doveadm_connection *client, const char *const *args)
 {
-       struct replicator_user *const *users, *user;
-       unsigned int i, count;
+       struct replicator_queue_iter *iter;
+       struct replicator_user *user;
        const char *mask = args[0];
        string_t *str = t_str_new(128);
 
        if (mask == NULL)
                return client_input_status_overview(client);
 
-       users = replicator_queue_get_users(client->queue, &count);
-       for (i = 0; i < count; i++) {
-               user = users[i];
+       iter = replicator_queue_iter_init(client->queue);
+       while ((user = replicator_queue_iter_next(iter)) != NULL) {
                if (!wildcard_match(user->username, mask))
                        continue;
 
@@ -94,6 +97,7 @@ client_input_status(struct doveadm_connection *client, const char *const *args)
                            user->last_sync_failed);
                o_stream_send(client->conn.output, str_data(str), str_len(str));
        }
+       replicator_queue_iter_deinit(&iter);
        o_stream_send(client->conn.output, "\n", 1);
        return 0;
 }
@@ -101,10 +105,11 @@ client_input_status(struct doveadm_connection *client, const char *const *args)
 static int
 client_input_replicate(struct doveadm_connection *client, const char *const *args)
 {
-       struct replicator_user *const *queue_users, **users_dup;
-       unsigned int i, count, match_count;
+       struct replicator_queue_iter *iter;
+       struct replicator_user *user;
        const char *usermask;
        enum replication_priority priority;
+       unsigned int match_count;
 
        /* <priority> <username>|<mask> */
        if (str_array_length(args) != 2) {
@@ -122,16 +127,15 @@ client_input_replicate(struct doveadm_connection *client, const char *const *arg
                return 0;
        }
 
-       queue_users = replicator_queue_get_users(client->queue, &count);
-       users_dup = i_new(struct replicator_user *, count+1);
-       for (i = match_count = 0; i < count; i++) {
-               if (wildcard_match(queue_users[i]->username, usermask))
-                       users_dup[match_count++] = queue_users[i];
-       }
-       for (i = 0; i < match_count; i++) {
-               replicator_queue_add(client->queue, users_dup[i]->username,
-                                    priority);
+       match_count = 0;
+       iter = replicator_queue_iter_init(client->queue);
+       while ((user = replicator_queue_iter_next(iter)) != NULL) {
+               if (!wildcard_match(user->username, usermask))
+                       continue;
+               replicator_queue_add(client->queue, user->username, priority);
+               match_count++;
        }
+       replicator_queue_iter_deinit(&iter);
        o_stream_send_str(client->conn.output,
                          t_strdup_printf("+%u\n", match_count));
        return 0;

diff --git a/src/replication/replicator/replicator-queue.c b/src/replication/replicator/replicator-queue.c
index 8b4c2a0850438eaea029f3a09fda82f857b77dbc..efbc86fb48a3f4d3dcb33bdc509794e1c8aab821 100644 (file)
--- a/src/replication/replicator/replicator-queue.c
+++ b/src/replication/replicator/replicator-queue.c
@@ -36,6 +36,11 @@ struct replicator_queue {
        void *change_context;
 };
 
+struct replicator_queue_iter {
+       struct replicator_queue *queue;
+       struct hash_iterate_context *iter;
+};
+
 static int user_priority_cmp(const void *p1, const void *p2)
 {
        const struct replicator_user *user1 = p1, *user2 = p2;
@@ -387,9 +392,9 @@ replicator_queue_export_user(struct replicator_user *user, string_t *str)
 
 int replicator_queue_export(struct replicator_queue *queue, const char *path)
 {
+       struct replicator_queue_iter *iter;
+       struct replicator_user *user;
        struct ostream *output;
-       struct priorityq_item *const *items;
-       unsigned int i, count;
        string_t *str;
        int fd, ret = 0;
 
@@ -402,17 +407,14 @@ int replicator_queue_export(struct replicator_queue *queue, const char *path)
        o_stream_cork(output);
 
        str = t_str_new(128);
-       items = priorityq_items(queue->user_queue);
-       count = priorityq_count(queue->user_queue);
-       for (i = 0; i < count; i++) {
-               struct replicator_user *user =
-                       (struct replicator_user *)items[i];
-
+       iter = replicator_queue_iter_init(queue);
+       while ((user = replicator_queue_iter_next(iter)) != NULL) {
                str_truncate(str, 0);
                replicator_queue_export_user(user, str);
                if (o_stream_send(output, str_data(str), str_len(str)) < 0)
                        break;
        }
+       replicator_queue_iter_deinit(&iter);
        if (o_stream_nfinish(output) < 0) {
                i_error("write(%s) failed: %m", path);
                ret = -1;
@@ -421,13 +423,35 @@ int replicator_queue_export(struct replicator_queue *queue, const char *path)
        return ret;
 }
 
-struct replicator_user *const *
-replicator_queue_get_users(struct replicator_queue *queue,
-                          unsigned int *count_r)
+struct replicator_queue_iter *
+replicator_queue_iter_init(struct replicator_queue *queue)
 {
-       struct priorityq_item *const *items =
-               priorityq_items(queue->user_queue);
+       struct replicator_queue_iter *iter;
+
+       iter = i_new(struct replicator_queue_iter, 1);
+       iter->queue = queue;
+       iter->iter = hash_table_iterate_init(queue->user_hash);
+       return iter;
+}
+
+struct replicator_user *
+replicator_queue_iter_next(struct replicator_queue_iter *iter)
+{
+       struct replicator_user *user;
+       char *username;
+
+       if (!hash_table_iterate(iter->iter, iter->queue->user_hash,
+                               &username, &user))
+               return NULL;
+       return user;
+}
+
+void replicator_queue_iter_deinit(struct replicator_queue_iter **_iter)
+{
+       struct replicator_queue_iter *iter = *_iter;
+
+       *_iter = NULL;
 
-       *count_r = priorityq_count(queue->user_queue);
-       return (void *)items;
+       hash_table_iterate_deinit(&iter->iter);
+       i_free(iter);
 }

diff --git a/src/replication/replicator/replicator-queue.h b/src/replication/replicator/replicator-queue.h
index 6119438944987a0638665f7571df803fca01bce1..b9fea45dd20d6aa8620eeab9ba3fcb1c152abbb4 100644 (file)
--- a/src/replication/replicator/replicator-queue.h
+++ b/src/replication/replicator/replicator-queue.h
@@ -69,9 +69,11 @@ int replicator_queue_export(struct replicator_queue *queue, const char *path);
 bool replicator_queue_want_sync_now(struct replicator_queue *queue,
                                    struct replicator_user *user,
                                    unsigned int *next_secs_r);
-/* Returns an (unsorted) array of all users in the queue. */
-struct replicator_user *const *
-replicator_queue_get_users(struct replicator_queue *queue,
-                          unsigned int *count_r);
+/* Iterate through all users in the queue. */
+struct replicator_queue_iter *
+replicator_queue_iter_init(struct replicator_queue *queue);
+struct replicator_user *
+replicator_queue_iter_next(struct replicator_queue_iter *iter);
+void replicator_queue_iter_deinit(struct replicator_queue_iter **iter);
 
 #endif