3faa1040e5a3f9f35ffad29110216094ab2f5880Timo Sirainen/* Copyright (c) 2013-2018 Dovecot authors, see the included COPYING file */
8dec8eab222cb7a0b4ef5e066cb1d7dac8a526d1Timo Sirainen
8dec8eab222cb7a0b4ef5e066cb1d7dac8a526d1Timo Sirainen#include "lib.h"
8dec8eab222cb7a0b4ef5e066cb1d7dac8a526d1Timo Sirainen#include "array.h"
d3b29d4b61f1549244a7509b798be6f806cf7d4eTimo Sirainen#include "ioloop.h"
f670747b681262e0bf004107c3b1d1fe1183d6ecTimo Sirainen#include "dsync-client.h"
85bc1cc40ba4566cb3cea251832800d0940908d5Aki Tuomi#include "replicator-settings.h"
85bc1cc40ba4566cb3cea251832800d0940908d5Aki Tuomi#include "replicator-queue.h"
d3b29d4b61f1549244a7509b798be6f806cf7d4eTimo Sirainen#include "replicator-brain.h"
f670747b681262e0bf004107c3b1d1fe1183d6ecTimo Sirainen
d3b29d4b61f1549244a7509b798be6f806cf7d4eTimo Sirainenstruct replicator_sync_context {
struct replicator_brain *brain;
struct replicator_user *user;
};
struct replicator_brain {
pool_t pool;
struct replicator_queue *queue;
const struct replicator_settings *set;
struct timeout *to;
ARRAY_TYPE(dsync_client) dsync_clients;
bool deinitializing:1;
};
static void replicator_brain_fill(struct replicator_brain *brain);
static void replicator_brain_queue_changed(void *context)
{
struct replicator_brain *brain = context;
replicator_brain_fill(brain);
}
struct replicator_brain *
replicator_brain_init(struct replicator_queue *queue,
const struct replicator_settings *set)
{
struct replicator_brain *brain;
pool_t pool;
pool = pool_alloconly_create("replication brain", 1024);
brain = p_new(pool, struct replicator_brain, 1);
brain->pool = pool;
brain->queue = queue;
brain->set = set;
p_array_init(&brain->dsync_clients, pool, 16);
replicator_queue_set_change_callback(queue,
replicator_brain_queue_changed, brain);
replicator_brain_fill(brain);
return brain;
}
void replicator_brain_deinit(struct replicator_brain **_brain)
{
struct replicator_brain *brain = *_brain;
struct dsync_client **connp;
*_brain = NULL;
brain->deinitializing = TRUE;
array_foreach_modifiable(&brain->dsync_clients, connp)
dsync_client_deinit(connp);
timeout_remove(&brain->to);
pool_unref(&brain->pool);
}
struct replicator_queue *
replicator_brain_get_queue(struct replicator_brain *brain)
{
return brain->queue;
}
const struct replicator_settings *
replicator_brain_get_settings(struct replicator_brain *brain)
{
return brain->set;
}
const ARRAY_TYPE(dsync_client) *
replicator_brain_get_dsync_clients(struct replicator_brain *brain)
{
return &brain->dsync_clients;
}
static struct dsync_client *
get_dsync_client(struct replicator_brain *brain)
{
struct dsync_client *const *connp, *conn = NULL;
array_foreach(&brain->dsync_clients, connp) {
if (!dsync_client_is_busy(*connp))
return *connp;
}
if (array_count(&brain->dsync_clients) ==
brain->set->replication_max_conns)
return NULL;
conn = dsync_client_init(brain->set->doveadm_socket_path,
brain->set->replication_dsync_parameters);
array_append(&brain->dsync_clients, &conn, 1);
return conn;
}
static void dsync_callback(enum dsync_reply reply, const char *state,
void *context)
{
struct replicator_sync_context *ctx = context;
struct replicator_user *user = ctx->user;
if (!replicator_user_unref(&user)) {
/* user was already removed */
} else if (reply == DSYNC_REPLY_NOUSER ||
reply == DSYNC_REPLY_NOREPLICATE) {
/* user no longer exists, or is not wanted for replication,
remove from replication */
replicator_queue_remove(ctx->brain->queue, &ctx->user);
} else {
i_free(ctx->user->state);
ctx->user->state = i_strdup_empty(state);
ctx->user->last_sync_failed = reply != DSYNC_REPLY_OK;
if (reply == DSYNC_REPLY_OK)
ctx->user->last_successful_sync = ioloop_time;
replicator_queue_push(ctx->brain->queue, ctx->user);
}
if (!ctx->brain->deinitializing)
replicator_brain_fill(ctx->brain);
i_free(ctx);
}
static bool
dsync_replicate(struct replicator_brain *brain, struct replicator_user *user)
{
struct replicator_sync_context *ctx;
struct dsync_client *conn;
time_t next_full_sync;
bool full;
conn = get_dsync_client(brain);
if (conn == NULL)
return FALSE;
next_full_sync = user->last_full_sync +
brain->set->replication_full_sync_interval;
full = next_full_sync <= ioloop_time;
/* update the sync times immediately. if the replication fails we still
wouldn't want it to be retried immediately. */
user->last_fast_sync = ioloop_time;
if (full || user->force_full_sync) {
user->last_full_sync = ioloop_time;
user->force_full_sync = FALSE;
}
/* reset priority also. if more updates arrive during replication
we'll do another replication to make sure nothing gets lost */
user->priority = REPLICATION_PRIORITY_NONE;
ctx = i_new(struct replicator_sync_context, 1);
ctx->brain = brain;
ctx->user = user;
replicator_user_ref(user);
dsync_client_sync(conn, user->username, user->state, full,
dsync_callback, ctx);
return TRUE;
}
static void replicator_brain_timeout(struct replicator_brain *brain)
{
timeout_remove(&brain->to);
replicator_brain_fill(brain);
}
static bool replicator_brain_fill_next(struct replicator_brain *brain)
{
struct replicator_user *user;
unsigned int next_secs;
user = replicator_queue_pop(brain->queue, &next_secs);
if (user == NULL) {
/* nothing more to do */
timeout_remove(&brain->to);
brain->to = timeout_add(next_secs * 1000,
replicator_brain_timeout, brain);
return FALSE;
}
if (!dsync_replicate(brain, user)) {
/* all connections were full, put the user back to queue */
replicator_queue_push(brain->queue, user);
return FALSE;
}
/* replication started for the user */
return TRUE;
}
static void replicator_brain_fill(struct replicator_brain *brain)
{
while (replicator_brain_fill_next(brain)) ;
}