replication/replicator/replicator-queue.c

	replicator-queue.c revision a8c5a86d183db25a57bf193c06b41e092ec2e151
/* Copyright (c) 2013-2014 Dovecot authors, see the included COPYING file */

#include "lib.h"
#include "array.h"
#include "ioloop.h"
#include "istream.h"
#include "ostream.h"
#include "str.h"
#include "strescape.h"
#include "hash.h"
#include "replicator-queue.h"

#include <unistd.h>
#include <fcntl.h>

struct replicator_sync_lookup {
    struct replicator_user *user;

    replicator_sync_callback_t *callback;
    void *context;

    bool wait_for_next_push;
};

struct replicator_queue {
    struct priorityq *user_queue;
    /* username => struct replicator_user* */
    HASH_TABLE(char *, struct replicator_user *) user_hash;

    ARRAY(struct replicator_sync_lookup) sync_lookups;

    unsigned int full_sync_interval;
    unsigned int failure_resync_interval;

    void (*change_callback)(void *context);
    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;

    if (user1->priority > user2->priority)
        return -1;
    if (user1->priority < user2->priority)
        return 1;

    if (user1->priority != REPLICATION_PRIORITY_NONE) {
        /* there is something to replicate */
        if (user1->last_fast_sync < user2->last_fast_sync)
            return -1;
        if (user1->last_fast_sync > user2->last_fast_sync)
            return 1;
    } else if (user1->last_sync_failed != user2->last_sync_failed) {
        /* resync failures first */
        if (user1->last_sync_failed)
            return -1;
        else
            return 1;
    } else if (user1->last_sync_failed) {
        /* both have failed. resync failures with fast-sync timestamp */
        if (user1->last_fast_sync < user2->last_fast_sync)
            return -1;
        if (user1->last_fast_sync > user2->last_fast_sync)
            return 1;
    } else {
        /* nothing to replicate, but do still periodic full syncs */
        if (user1->last_full_sync < user2->last_full_sync)
            return -1;
        if (user1->last_full_sync > user2->last_full_sync)
            return 1;
    }
    return 0;
}

struct replicator_queue *
replicator_queue_init(unsigned int full_sync_interval,
              unsigned int failure_resync_interval)
{
    struct replicator_queue *queue;

    queue = i_new(struct replicator_queue, 1);
    queue->full_sync_interval = full_sync_interval;
    queue->failure_resync_interval = failure_resync_interval;
    queue->user_queue = priorityq_init(user_priority_cmp, 1024);
    hash_table_create(&queue->user_hash, default_pool, 1024,
              str_hash, strcmp);
    i_array_init(&queue->sync_lookups, 32);
    return queue;
}

void replicator_queue_deinit(struct replicator_queue **_queue)
{
    struct replicator_queue *queue = *_queue;
    struct priorityq_item *item;

    *_queue = NULL;

    queue->change_callback = NULL;

    while ((item = priorityq_pop(queue->user_queue)) != NULL) {
        struct replicator_user *user = (struct replicator_user *)item;

        user->popped = TRUE;
        replicator_queue_remove(queue, &user);
    }

    priorityq_deinit(&queue->user_queue);
    hash_table_destroy(&queue->user_hash);
    i_assert(array_count(&queue->sync_lookups) == 0);
    array_free(&queue->sync_lookups);
    i_free(queue);
}

void replicator_queue_set_change_callback(struct replicator_queue *queue,
                      void (*callback)(void *context),
                      void *context)
{
    queue->change_callback = callback;
    queue->change_context = context;
}

struct replicator_user *
replicator_queue_lookup(struct replicator_queue *queue, const char *username)
{
    return hash_table_lookup(queue->user_hash, username);
}

static struct replicator_user *
replicator_queue_add_int(struct replicator_queue *queue, const char *username,
             enum replication_priority priority)
{
    struct replicator_user *user;

    user = replicator_queue_lookup(queue, username);
    if (user == NULL) {
        user = i_new(struct replicator_user, 1);
        user->username = i_strdup(username);
        hash_table_insert(queue->user_hash, user->username, user);
    } else {
        if (user->priority > priority) {
            /* user already has a higher priority than this */
            return user;
        }
        if (!user->popped)
            priorityq_remove(queue->user_queue, &user->item);
    }
    user->priority = priority;
    user->last_update = ioloop_time;

    if (!user->popped)
        priorityq_add(queue->user_queue, &user->item);
    return user;
}

struct replicator_user *
replicator_queue_add(struct replicator_queue *queue, const char *username,
             enum replication_priority priority)
{
    struct replicator_user *user;

    user = replicator_queue_add_int(queue, username, priority);
    if (queue->change_callback != NULL)
        queue->change_callback(queue->change_context);
    return user;
}

void replicator_queue_add_sync(struct replicator_queue *queue,
                   const char *username,
                   replicator_sync_callback_t *callback,
                   void *context)
{
    struct replicator_user *user;
    struct replicator_sync_lookup *lookup;

    user = replicator_queue_add_int(queue, username,
                    REPLICATION_PRIORITY_SYNC);

    lookup = array_append_space(&queue->sync_lookups);
    lookup->user = user;
    lookup->callback = callback;
    lookup->context = context;
    lookup->wait_for_next_push = user->popped;

    if (queue->change_callback != NULL)
        queue->change_callback(queue->change_context);
}

void replicator_queue_remove(struct replicator_queue *queue,
                 struct replicator_user **_user)
{
    struct replicator_user *user = *_user;

    *_user = NULL;
    if (!user->popped)
        priorityq_remove(queue->user_queue, &user->item);
    hash_table_remove(queue->user_hash, user->username);

    i_free(user->state);
    i_free(user->username);
    i_free(user);

    if (queue->change_callback != NULL)
        queue->change_callback(queue->change_context);
}

bool replicator_queue_want_sync_now(struct replicator_queue *queue,
                    struct replicator_user *user,
                    unsigned int *next_secs_r)
{
    time_t next_sync;

    if (user->priority != REPLICATION_PRIORITY_NONE)
        return TRUE;

    if (user->last_sync_failed) {
        next_sync = user->last_fast_sync +
            queue->failure_resync_interval;
    } else {
        next_sync = user->last_full_sync + queue->full_sync_interval;
    }
    if (next_sync <= ioloop_time)
        return TRUE;

    *next_secs_r = next_sync - ioloop_time;
    return FALSE;
}

struct replicator_user *
replicator_queue_pop(struct replicator_queue *queue,
             unsigned int *next_secs_r)
{
    struct priorityq_item *item;
    struct replicator_user *user;

    item = priorityq_peek(queue->user_queue);
    if (item == NULL) {
        /* no users defined. we shouldn't normally get here */
        *next_secs_r = 3600;
        return NULL;
    }
    user = (struct replicator_user *)item;
    if (!replicator_queue_want_sync_now(queue, user, next_secs_r)) {
        /* we don't want to sync the user yet */
        return NULL;
    }
    priorityq_remove(queue->user_queue, &user->item);
    user->popped = TRUE;
    return user;
}

static void
replicator_queue_handle_sync_lookups(struct replicator_queue *queue,
                     struct replicator_user *user)
{
    struct replicator_sync_lookup *lookups;
    ARRAY(struct replicator_sync_lookup) callbacks;
    unsigned int i, count;
    bool success = !user->last_sync_failed;

    t_array_init(&callbacks, 8);
    lookups = array_get_modifiable(&queue->sync_lookups, &count);
    for (i = 0; i < count; ) {
        if (lookups[i].user != user)
            i++;
        else if (lookups[i].wait_for_next_push) {
            /* another sync request came while user was being
               replicated */
            i_assert(user->priority == REPLICATION_PRIORITY_SYNC);
            lookups[i].wait_for_next_push = FALSE;
            i++;
        } else {
            array_append(&callbacks, &lookups[i], 1);
            array_delete(&queue->sync_lookups, i, 1);
        }
    }

    array_foreach_modifiable(&callbacks, lookups)
        lookups->callback(success, lookups->context);
}

void replicator_queue_push(struct replicator_queue *queue,
               struct replicator_user *user)
{
    i_assert(user->popped);

    priorityq_add(queue->user_queue, &user->item);
    user->popped = FALSE;

    T_BEGIN {
        replicator_queue_handle_sync_lookups(queue, user);
    } T_END;
}

static int
replicator_queue_import_line(struct replicator_queue *queue, const char *line)
{
    const char *const *args, *username, *state;
    unsigned int priority;
    struct replicator_user *user, tmp_user;

    /* <user> <priority> <last update> <last fast sync> <last full sync>
       <last failed> <state> */
    args = t_strsplit_tabescaped(line);
    if (str_array_length(args) < 7)
        return -1;

    memset(&tmp_user, 0, sizeof(tmp_user));
    username = args[0];
    state = t_strdup_noconst(args[6]);
    if (username[0] == '\0' ||
        str_to_uint(args[1], &priority) < 0 ||
        str_to_time(args[2], &tmp_user.last_update) < 0 ||
        str_to_time(args[3], &tmp_user.last_fast_sync) < 0 ||
        str_to_time(args[4], &tmp_user.last_full_sync) < 0)
        return -1;
    tmp_user.priority = priority;
    tmp_user.last_sync_failed = args[5][0] != '0';

    user = hash_table_lookup(queue->user_hash, username);
    if (user != NULL) {
        if (user->last_update > tmp_user.last_update) {
            /* we already have a newer state */
            return 0;
        }
        if (user->last_update == tmp_user.last_update) {
            /* either one of these could be newer. use the one
               with higher priority. */
            if (user->priority > tmp_user.priority)
                return 0;
        }
    }
    user = replicator_queue_add(queue, username,
                    tmp_user.priority);
    user->last_update = tmp_user.last_update;
    user->last_fast_sync = tmp_user.last_fast_sync;
    user->last_full_sync = tmp_user.last_full_sync;
    user->last_sync_failed = tmp_user.last_sync_failed;
    i_free(user->state);
    user->state = i_strdup(state);
    return 0;
}

int replicator_queue_import(struct replicator_queue *queue, const char *path)
{
    struct istream *input;
    const char *line;
    int fd, ret = 0;

    fd = open(path, O_RDONLY);
    if (fd == -1) {
        if (errno == ENOENT)
            return 0;
        i_error("open(%s) failed: %m", path);
        return -1;
    }

    input = i_stream_create_fd(fd, (size_t)-1, TRUE);
    while ((line = i_stream_read_next_line(input)) != NULL) {
        T_BEGIN {
            ret = replicator_queue_import_line(queue, line);
        } T_END;
        if (ret < 0) {
            i_error("Corrupted replicator record in %s: %s",
                path, line);
            break;
        }
    }
    if (input->stream_errno != 0)
        ret = -1;
    i_stream_destroy(&input);
    return ret;
}

static void
replicator_queue_export_user(struct replicator_user *user, string_t *str)
{
    str_append_tabescaped(str, user->username);
    str_printfa(str, "\t%d\t%lld\t%lld\t%lld\t%d\t", (int)user->priority,
            (long long)user->last_update,
            (long long)user->last_fast_sync,
            (long long)user->last_full_sync,
            user->last_sync_failed);
    if (user->state != NULL)
        str_append_tabescaped(str, user->state);
    str_append_c(str, '\n');
}

int replicator_queue_export(struct replicator_queue *queue, const char *path)
{
    struct replicator_queue_iter *iter;
    struct replicator_user *user;
    struct ostream *output;
    string_t *str;
    int fd, ret = 0;

    fd = creat(path, 0600);
    if (fd == -1) {
        i_error("creat(%s) failed: %m", path);
        return -1;
    }
    output = o_stream_create_fd_file(fd, 0, TRUE);
    o_stream_cork(output);

    str = t_str_new(128);
    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;
    }
    o_stream_destroy(&output);
    return ret;
}

struct replicator_queue_iter *
replicator_queue_iter_init(struct replicator_queue *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;

    hash_table_iterate_deinit(&iter->iter);
    i_free(iter);
}