14944

pending completion

Build # 14944

Build Type

push

travis-ci

Committed by

locker

Commit Message

<a href="https://github.com/tarantool/tarantool/commit/<a class=hub.com/tarantool/tarantool/commit/f1170c438133c38ac13c36a268f268ebd192a477">f1170c438<a href="https://github.com/tarantool/tarantool/commit/f1170c438133c38ac13c36a268f268ebd192a477">&quot;&gt;session: outdate a session of a closed connection

Once a connection is closed, a long-running user
request can not learn about this occasion. Even
box.sesion.push() still works.

This patch makes such disconnected session &amp;#39;rotten&amp;#39;.
So a user can determine if a connection is closed by
looking at session.fd() == -1, or checking for
errors from box.session.push().

Closes #3859

(cherry picked from commit </a><a class="double-link" href="https://github.com/tarantool/tarantool/commit/<a class="double-link" href="https://github.com/tarantool/tarantool/commit/43af2de2b129dabafe94a61c89688dd4701f8800">43af2de2b</a>">43af2de2b</a><a href="https://github.com/tarantool/tarantool/commit/f1170c438133c38ac13c36a268f268ebd192a477">)

Run Details

12 of 12 new or added lines in 2 files covered. (100.0%)

41102 of 49943 relevant lines covered (82.3%)

843020.17 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.98

/src/box/applier.cc

/*
 * Copyright 2010-2016, Tarantool AUTHORS, please see AUTHORS file.
 *
 * Redistribution and use in source and binary forms, with or
 * without modification, are permitted provided that the following
 * conditions are met:
 *
 * 1. Redistributions of source code must retain the above
 *    copyright notice, this list of conditions and the
 *    following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above
 *    copyright notice, this list of conditions and the following
 *    disclaimer in the documentation and/or other materials
 *    provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
 * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
 * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
 * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
#include "applier.h"

#include <msgpuck.h>

#include "xlog.h"
#include "fiber.h"
#include "fiber_cond.h"
#include "coio.h"
#include "coio_buf.h"
#include "xstream.h"
#include "wal.h"
#include "xrow.h"
#include "replication.h"
#include "iproto_constants.h"
#include "version.h"
#include "trigger.h"
#include "xrow_io.h"
#include "error.h"
#include "session.h"
#include "cfg.h"

STRS(applier_state, applier_STATE);

static inline void
applier_set_state(struct applier *applier, enum applier_state state)
{
        applier->state = state;
        say_debug("=> %s", applier_state_strs[state] +
                  strlen("APPLIER_"));
        trigger_run_xc(&applier->on_state, applier);
}

/**
 * Write a nice error message to log file on SocketError or ClientError
 * in applier_f().
 */
static inline void
applier_log_error(struct applier *applier, struct error *e)
{
        uint32_t errcode = box_error_code(e);
        if (applier->last_logged_errcode == errcode)
                return;
        switch (applier->state) {
        case APPLIER_CONNECT:
                say_info("can't connect to master");
                break;
        case APPLIER_CONNECTED:
        case APPLIER_READY:
                say_info("can't join/subscribe");
                break;
        case APPLIER_AUTH:
                say_info("failed to authenticate");
                break;
        case APPLIER_SYNC:
        case APPLIER_FOLLOW:
        case APPLIER_INITIAL_JOIN:
        case APPLIER_FINAL_JOIN:
                say_info("can't read row");
                break;
        default:
                break;
        }
        error_log(e);
        switch (errcode) {
        case ER_LOADING:
        case ER_CFG:
        case ER_ACCESS_DENIED:
        case ER_NO_SUCH_USER:
        case ER_SYSTEM:
                say_info("will retry every %.2lf second",
                         replication_reconnect_interval());
                break;
        default:
                break;
        }
        applier->last_logged_errcode = errcode;
}

/*
 * Fiber function to write vclock to replication master.
 * To track connection status, replica answers master
 * with encoded vclock. In addition to DML requests,
 * master also sends heartbeat messages every
 * replication_timeout seconds (introduced in 1.7.7).
 * On such requests replica also responds with vclock.
 */
static int
applier_writer_f(va_list ap)
{
        struct applier *applier = va_arg(ap, struct applier *);
        struct ev_io io;
        coio_create(&io, applier->io.fd);

        while (!fiber_is_cancelled()) {
                /*
                 * Tarantool >= 1.7.7 sends periodic heartbeat
                 * messages so we don't need to send ACKs every
                 * replication_timeout seconds any more.
                 */
                if (applier->version_id >= version_id(1, 7, 7))
                        fiber_cond_wait_timeout(&applier->writer_cond,
                                                TIMEOUT_INFINITY);
                else
                        fiber_cond_wait_timeout(&applier->writer_cond,
                                                replication_timeout);
                /* Send ACKs only when in FOLLOW mode ,*/
                if (applier->state != APPLIER_SYNC &&
                    applier->state != APPLIER_FOLLOW)
                        continue;
                try {
                        struct xrow_header xrow;
                        xrow_encode_vclock(&xrow, &replicaset.vclock);
                        coio_write_xrow(&io, &xrow);
                } catch (SocketError *e) {
                        /*
                         * There is no point trying to send ACKs if
                         * the master closed its end - we would only
                         * spam the log - so exit immediately.
                         */
                        if (e->get_errno() == EPIPE)
                                break;
                        /*
                         * Do not exit, if there is a network error,
                         * the reader fiber will reconnect for us
                         * and signal our cond afterwards.
                         */
                        e->log();
                } catch (Exception *e) {
                        /*
                         * Out of memory encoding the message, ignore
                         * and try again after an interval.
                         */
                        e->log();
                }
                fiber_gc();
        }
        return 0;
}

/**
 * Connect to a remote host and authenticate the client.
 */
void
applier_connect(struct applier *applier)
{
        struct ev_io *coio = &applier->io;
        struct ibuf *ibuf = &applier->ibuf;
        if (coio->fd >= 0)
                return;
        char greetingbuf[IPROTO_GREETING_SIZE];
        struct xrow_header row;

        struct uri *uri = &applier->uri;
        /*
         * coio_connect() stores resolved address to \a &applier->addr
         * on success. &applier->addr_len is a value-result argument which
         * must be initialized to the size of associated buffer (addrstorage)
         * before calling coio_connect(). Since coio_connect() performs
         * DNS resolution under the hood it is theoretically possible that
         * applier->addr_len will be different even for same uri.
         */
        applier->addr_len = sizeof(applier->addrstorage);
        applier_set_state(applier, APPLIER_CONNECT);
        coio_connect(coio, uri, &applier->addr, &applier->addr_len);
        assert(coio->fd >= 0);
        coio_readn(coio, greetingbuf, IPROTO_GREETING_SIZE);
        applier->last_row_time = ev_monotonic_now(loop());

        /* Decode instance version and name from greeting */
        struct greeting greeting;
        if (greeting_decode(greetingbuf, &greeting) != 0)
                tnt_raise(LoggedError, ER_PROTOCOL, "Invalid greeting");

        if (strcmp(greeting.protocol, "Binary") != 0) {
                tnt_raise(LoggedError, ER_PROTOCOL,
                          "Unsupported protocol for replication");
        }

        if (applier->version_id != greeting.version_id) {
                say_info("remote master is %u.%u.%u at %s",
                         version_id_major(greeting.version_id),
                         version_id_minor(greeting.version_id),
                         version_id_patch(greeting.version_id),
                         sio_strfaddr(&applier->addr, applier->addr_len));
        }

        /* Save the remote instance version and UUID on connect. */
        applier->uuid = greeting.uuid;
        applier->version_id = greeting.version_id;

        /* Don't display previous error messages in box.info.replication */
        diag_clear(&fiber()->diag);

        /*
         * Send an IPROTO_VOTE request to fetch the master's ballot
         * before proceeding to "join". It will be used for leader
         * election on bootstrap.
         */
        xrow_encode_vote(&row);
        coio_write_xrow(coio, &row);
        coio_read_xrow(coio, ibuf, &row);
        if (row.type == IPROTO_OK) {
                xrow_decode_ballot_xc(&row, &applier->ballot);
        } else try {
                xrow_decode_error_xc(&row);
        } catch (ClientError *e) {
                if (e->errcode() != ER_UNKNOWN_REQUEST_TYPE)
                        e->raise();
                /*
                 * Master isn't aware of IPROTO_VOTE request.
                 * It's OK - we can proceed without it.
                 */
        }

        applier_set_state(applier, APPLIER_CONNECTED);

        /* Detect connection to itself */
        if (tt_uuid_is_equal(&applier->uuid, &INSTANCE_UUID))
                tnt_raise(ClientError, ER_CONNECTION_TO_SELF);

        /* Perform authentication if user provided at least login */
        if (!uri->login)
                goto done;

        /* Authenticate */
        applier_set_state(applier, APPLIER_AUTH);
        xrow_encode_auth_xc(&row, greeting.salt, greeting.salt_len, uri->login,
                            uri->login_len, uri->password, uri->password_len);
        coio_write_xrow(coio, &row);
        coio_read_xrow(coio, ibuf, &row);
        applier->last_row_time = ev_monotonic_now(loop());
        if (row.type != IPROTO_OK)
                xrow_decode_error_xc(&row); /* auth failed */

        /* auth succeeded */
        say_info("authenticated");
done:
        applier_set_state(applier, APPLIER_READY);
}

/**
 * Execute and process JOIN request (bootstrap the instance).
 */
static void
applier_join(struct applier *applier)
{
        /* Send JOIN request */
        struct ev_io *coio = &applier->io;
        struct ibuf *ibuf = &applier->ibuf;
        struct xrow_header row;
        xrow_encode_join_xc(&row, &INSTANCE_UUID);
        coio_write_xrow(coio, &row);

        /**
         * Tarantool < 1.7.0: if JOIN is successful, there is no "OK"
         * response, but a stream of rows from checkpoint.
         */
        if (applier->version_id >= version_id(1, 7, 0)) {
                /* Decode JOIN response */
                coio_read_xrow(coio, ibuf, &row);
                if (iproto_type_is_error(row.type)) {
                        xrow_decode_error_xc(&row); /* re-throw error */
                } else if (row.type != IPROTO_OK) {
                        tnt_raise(ClientError, ER_UNKNOWN_REQUEST_TYPE,
                                  (uint32_t) row.type);
                }
                /*
                 * Start vclock. The vclock of the checkpoint
                 * the master is sending to the replica.
                 * Used to initialize the replica's initial
                 * vclock in bootstrap_from_master()
                 */
                xrow_decode_vclock_xc(&row, &replicaset.vclock);
        }

        applier_set_state(applier, APPLIER_INITIAL_JOIN);

        /*
         * Receive initial data.
         */
        assert(applier->join_stream != NULL);
        uint64_t row_count = 0;
        while (true) {
                coio_read_xrow(coio, ibuf, &row);
                applier->last_row_time = ev_monotonic_now(loop());
                if (iproto_type_is_dml(row.type)) {
                        xstream_write_xc(applier->join_stream, &row);
                        if (++row_count % 100000 == 0)
                                say_info("%.1fM rows received", row_count / 1e6);
                } else if (row.type == IPROTO_OK) {
                        if (applier->version_id < version_id(1, 7, 0)) {
                                /*
                                 * This is the start vclock if the
                                 * server is 1.6. Since we have
                                 * not initialized replication
                                 * vclock yet, do it now. In 1.7+
                                 * this vclock is not used.
                                 */
                                xrow_decode_vclock_xc(&row, &replicaset.vclock);
                        }
                        break; /* end of stream */
                } else if (iproto_type_is_error(row.type)) {
                        xrow_decode_error_xc(&row);  /* rethrow error */
                } else {
                        tnt_raise(ClientError, ER_UNKNOWN_REQUEST_TYPE,
                                  (uint32_t) row.type);
                }
        }
        say_info("initial data received");

        applier_set_state(applier, APPLIER_FINAL_JOIN);

        /*
         * Tarantool < 1.7.0: there is no "final join" stage.
         * Proceed to "subscribe" and do not finish bootstrap
         * until replica id is received.
         */
        if (applier->version_id < version_id(1, 7, 0))
                return;

        /*
         * Receive final data.
         */
        while (true) {
                coio_read_xrow(coio, ibuf, &row);
                applier->last_row_time = ev_monotonic_now(loop());
                if (iproto_type_is_dml(row.type)) {
                        vclock_follow_xrow(&replicaset.vclock, &row);
                        xstream_write_xc(applier->subscribe_stream, &row);
                        if (++row_count % 100000 == 0)
                                say_info("%.1fM rows received", row_count / 1e6);
                } else if (row.type == IPROTO_OK) {
                        /*
                         * Current vclock. This is not used now,
                         * ignore.
                         */
                        break; /* end of stream */
                } else if (iproto_type_is_error(row.type)) {
                        xrow_decode_error_xc(&row);  /* rethrow error */
                } else {
                        tnt_raise(ClientError, ER_UNKNOWN_REQUEST_TYPE,
                                  (uint32_t) row.type);
                }
        }
        say_info("final data received");

        applier_set_state(applier, APPLIER_JOINED);
        applier_set_state(applier, APPLIER_READY);
}

/**
 * Execute and process SUBSCRIBE request (follow updates from a master).
 */
static void
applier_subscribe(struct applier *applier)
{
        assert(applier->subscribe_stream != NULL);

        /* Send SUBSCRIBE request */
        struct ev_io *coio = &applier->io;
        struct ibuf *ibuf = &applier->ibuf;
        struct xrow_header row;
        struct vclock remote_vclock_at_subscribe;

        xrow_encode_subscribe_xc(&row, &REPLICASET_UUID, &INSTANCE_UUID,
                                 &replicaset.vclock);
        coio_write_xrow(coio, &row);

        /* Read SUBSCRIBE response */
        if (applier->version_id >= version_id(1, 6, 7)) {
                coio_read_xrow(coio, ibuf, &row);
                if (iproto_type_is_error(row.type)) {
                        xrow_decode_error_xc(&row);  /* error */
                } else if (row.type != IPROTO_OK) {
                        tnt_raise(ClientError, ER_PROTOCOL,
                                  "Invalid response to SUBSCRIBE");
                }
                /*
                 * In case of successful subscribe, the server
                 * responds with its current vclock.
                 */
                vclock_create(&remote_vclock_at_subscribe);
                xrow_decode_vclock_xc(&row, &remote_vclock_at_subscribe);
        }
        /*
         * Tarantool < 1.6.7:
         * If there is an error in subscribe, it's sent directly
         * in response to subscribe.  If subscribe is successful,
         * there is no "OK" response, but a stream of rows from
         * the binary log.
         */

        if (applier->state == APPLIER_READY) {
                /*
                 * Tarantool < 1.7.7 does not send periodic heartbeat
                 * messages so we cannot enable applier synchronization
                 * for it without risking getting stuck in the 'orphan'
                 * mode until a DML operation happens on the master.
                 */
                if (applier->version_id >= version_id(1, 7, 7))
                        applier_set_state(applier, APPLIER_SYNC);
                else
                        applier_set_state(applier, APPLIER_FOLLOW);
        } else {
                /*
                 * Tarantool < 1.7.0 sends replica id during
                 * "subscribe" stage. We can't finish bootstrap
                 * until it is received.
                 */
                assert(applier->state == APPLIER_FINAL_JOIN);
                assert(applier->version_id < version_id(1, 7, 0));
        }

        /* Re-enable warnings after successful execution of SUBSCRIBE */
        applier->last_logged_errcode = 0;
        if (applier->version_id >= version_id(1, 7, 4)) {
                /* Enable replication ACKs for newer servers */
                assert(applier->writer == NULL);

                char name[FIBER_NAME_MAX];
                int pos = snprintf(name, sizeof(name), "applierw/");
                uri_format(name + pos, sizeof(name) - pos, &applier->uri, false);

                applier->writer = fiber_new_xc(name, applier_writer_f);
                fiber_set_joinable(applier->writer, true);
                fiber_start(applier->writer, applier);
        }

        applier->lag = TIMEOUT_INFINITY;

        /*
         * Process a stream of rows from the binary log.
         */
        while (true) {
                if (applier->state == APPLIER_FINAL_JOIN &&
                    instance_id != REPLICA_ID_NIL) {
                        say_info("final data received");
                        applier_set_state(applier, APPLIER_JOINED);
                        applier_set_state(applier, APPLIER_READY);
                        applier_set_state(applier, APPLIER_FOLLOW);
                }

                /*
                 * Stay 'orphan' until appliers catch up with
                 * the remote vclock at the time of SUBSCRIBE
                 * and the lag is less than configured.
                 */
                if (applier->state == APPLIER_SYNC &&
                    applier->lag <= replication_sync_lag &&
                    vclock_compare(&remote_vclock_at_subscribe,
                                   &replicaset.vclock) <= 0) {
                        /* Applier is synced, switch to "follow". */
                        applier_set_state(applier, APPLIER_FOLLOW);
                }

                /*
                 * Tarantool < 1.7.7 does not send periodic heartbeat
                 * messages so we can't assume that if we haven't heard
                 * from the master for quite a while the connection is
                 * broken - the master might just be idle.
                 */
                if (applier->version_id < version_id(1, 7, 7)) {
                        coio_read_xrow(coio, ibuf, &row);
                } else {
                        double timeout = replication_disconnect_timeout();
                        coio_read_xrow_timeout_xc(coio, ibuf, &row, timeout);
                }

                if (iproto_type_is_error(row.type))
                        xrow_decode_error_xc(&row);  /* error */
                /* Replication request. */
                if (row.replica_id == REPLICA_ID_NIL ||
                    row.replica_id >= VCLOCK_MAX) {
                        /*
                         * A safety net, this can only occur
                         * if we're fed a strangely broken xlog.
                         */
                        tnt_raise(ClientError, ER_UNKNOWN_REPLICA,
                                  int2str(row.replica_id),
                                  tt_uuid_str(&REPLICASET_UUID));
                }

                applier->lag = ev_now(loop()) - row.tm;
                applier->last_row_time = ev_monotonic_now(loop());

                if (vclock_get(&replicaset.vclock, row.replica_id) < row.lsn) {
                        /**
                         * Promote the replica set vclock before
                         * applying the row. If there is an
                         * exception (conflict) applying the row,
                         * the row is skipped when the replication
                         * is resumed.
                         */
                        vclock_follow_xrow(&replicaset.vclock, &row);
                        struct replica *replica = replica_by_id(row.replica_id);
                        struct latch *latch = (replica ? &replica->order_latch :
                                               &replicaset.applier.order_latch);
                        /*
                         * In a full mesh topology, the same set
                         * of changes may arrive via two
                         * concurrently running appliers. Thanks
                         * to vclock_follow() above, the first row
                         * in the set will be skipped - but the
                         * remaining may execute out of order,
                         * when the following xstream_write()
                         * yields on WAL. Hence we need a latch to
                         * strictly order all changes which belong
                         * to the same server id.
                         */
                        latch_lock(latch);
                        int res = xstream_write(applier->subscribe_stream, &row);
                        latch_unlock(latch);
                        if (res != 0) {
                                struct error *e = diag_last_error(diag_get());
                                /**
                                 * Silently skip ER_TUPLE_FOUND error if such
                                 * option is set in config.
                                 */
                                if (e->type == &type_ClientError &&
                                    box_error_code(e) == ER_TUPLE_FOUND &&
                                    replication_skip_conflict)
                                        diag_clear(diag_get());
                                else
                                        diag_raise();
                        }
                }
                if (applier->state == APPLIER_SYNC ||
                    applier->state == APPLIER_FOLLOW)
                        fiber_cond_signal(&applier->writer_cond);
                if (ibuf_used(ibuf) == 0)
                        ibuf_reset(ibuf);
                fiber_gc();
        }
}

static inline void
applier_disconnect(struct applier *applier, enum applier_state state)
{
        applier_set_state(applier, state);
        if (applier->writer != NULL) {
                fiber_cancel(applier->writer);
                fiber_join(applier->writer);
                applier->writer = NULL;
        }

        coio_close(loop(), &applier->io);
        /* Clear all unparsed input. */
        ibuf_reinit(&applier->ibuf);
        fiber_gc();
}

static int
applier_f(va_list ap)
{
        struct applier *applier = va_arg(ap, struct applier *);
        /*
         * Set correct session type for use in on_replace()
         * triggers.
         */
        struct session *session = session_create_on_demand();
        if (session == NULL)
                return -1;
        session_set_type(session, SESSION_TYPE_APPLIER);

        /* Re-connect loop */
        while (!fiber_is_cancelled()) {
                try {
                        applier_connect(applier);
                        if (tt_uuid_is_nil(&REPLICASET_UUID)) {
                                /*
                                 * Execute JOIN if this is a bootstrap.
                                 * The join will pause the applier
                                 * until WAL is created.
                                 */
                                applier_join(applier);
                        }
                        applier_subscribe(applier);
                        /*
                         * subscribe() has an infinite loop which
                         * is stoppable only with fiber_cancel().
                         */
                        unreachable();
                        return 0;
                } catch (ClientError *e) {
                        if (e->errcode() == ER_CONNECTION_TO_SELF &&
                            tt_uuid_is_equal(&applier->uuid, &INSTANCE_UUID)) {
                                /* Connection to itself, stop applier */
                                applier_disconnect(applier, APPLIER_OFF);
                                return 0;
                        } else if (e->errcode() == ER_LOADING) {
                                /* Autobootstrap */
                                applier_log_error(applier, e);
                                applier_disconnect(applier, APPLIER_LOADING);
                                goto reconnect;
                        } else if (e->errcode() == ER_CFG ||
                                   e->errcode() == ER_ACCESS_DENIED ||
                                   e->errcode() == ER_NO_SUCH_USER) {
                                /* Invalid configuration */
                                applier_log_error(applier, e);
                                applier_disconnect(applier, APPLIER_LOADING);
                                goto reconnect;
                        } else if (e->errcode() == ER_SYSTEM) {
                                /* System error from master instance. */
                                applier_log_error(applier, e);
                                applier_disconnect(applier, APPLIER_DISCONNECTED);
                                goto reconnect;
                        } else {
                                /* Unrecoverable errors */
                                applier_log_error(applier, e);
                                applier_disconnect(applier, APPLIER_STOPPED);
                                return -1;
                        }
                } catch (FiberIsCancelled *e) {
                        applier_disconnect(applier, APPLIER_OFF);
                        break;
                } catch (SocketError *e) {
                        applier_log_error(applier, e);
                        applier_disconnect(applier, APPLIER_DISCONNECTED);
                        goto reconnect;
                } catch (SystemError *e) {
                        applier_log_error(applier, e);
                        applier_disconnect(applier, APPLIER_DISCONNECTED);
                        goto reconnect;
                } catch (Exception *e) {
                        applier_log_error(applier, e);
                        applier_disconnect(applier, APPLIER_STOPPED);
                        return -1;
                }
                /* Put fiber_sleep() out of catch block.
                 *
                 * This is done to avoid the case when two or more
                 * fibers yield inside their try/catch blocks and
                 * throw an exception. Seems like the exception unwinder
                 * uses global state inside the catch block.
                 *
                 * This could lead to incorrect exception processing
                 * and crash the program.
                 *
                 * See: https://github.com/tarantool/tarantool/issues/136
                */
reconnect:
                fiber_sleep(replication_reconnect_interval());
        }
        return 0;
}

void
applier_start(struct applier *applier)
{
        char name[FIBER_NAME_MAX];
        assert(applier->reader == NULL);

        int pos = snprintf(name, sizeof(name), "applier/");
        uri_format(name + pos, sizeof(name) - pos, &applier->uri, false);

        struct fiber *f = fiber_new_xc(name, applier_f);
        /**
         * So that we can safely grab the status of the
         * fiber any time we want.
         */
        fiber_set_joinable(f, true);
        applier->reader = f;
        fiber_start(f, applier);
}

void
applier_stop(struct applier *applier)
{
        struct fiber *f = applier->reader;
        if (f == NULL)
                return;
        fiber_cancel(f);
        fiber_join(f);
        applier_set_state(applier, APPLIER_OFF);
        applier->reader = NULL;
}

struct applier *
applier_new(const char *uri, struct xstream *join_stream,
            struct xstream *subscribe_stream)
{
        struct applier *applier = (struct applier *)
                calloc(1, sizeof(struct applier));
        if (applier == NULL) {
                diag_set(OutOfMemory, sizeof(*applier), "malloc",
                         "struct applier");
                return NULL;
        }
        coio_create(&applier->io, -1);
        ibuf_create(&applier->ibuf, &cord()->slabc, 1024);

        /* uri_parse() sets pointers to applier->source buffer */
        snprintf(applier->source, sizeof(applier->source), "%s", uri);
        int rc = uri_parse(&applier->uri, applier->source);
        /* URI checked by box_check_replication() */
        assert(rc == 0 && applier->uri.service != NULL);
        (void) rc;

        applier->join_stream = join_stream;
        applier->subscribe_stream = subscribe_stream;
        applier->last_row_time = ev_monotonic_now(loop());
        rlist_create(&applier->on_state);
        fiber_cond_create(&applier->resume_cond);
        fiber_cond_create(&applier->writer_cond);

        return applier;
}

void
applier_delete(struct applier *applier)
{
        assert(applier->reader == NULL && applier->writer == NULL);
        ibuf_destroy(&applier->ibuf);
        assert(applier->io.fd == -1);
        trigger_destroy(&applier->on_state);
        fiber_cond_destroy(&applier->resume_cond);
        fiber_cond_destroy(&applier->writer_cond);
        free(applier);
}

void
applier_resume(struct applier *applier)
{
        assert(!fiber_is_dead(applier->reader));
        applier->is_paused = false;
        fiber_cond_signal(&applier->resume_cond);
}

void
applier_pause(struct applier *applier)
{
        /* Sleep until applier_resume() wake us up */
        assert(fiber() == applier->reader);
        assert(!applier->is_paused);
        applier->is_paused = true;
        while (applier->is_paused && !fiber_is_cancelled())
                fiber_cond_wait(&applier->resume_cond);
}

struct applier_on_state {
        struct trigger base;
        struct applier *applier;
        enum applier_state desired_state;
        struct fiber_cond wakeup;
};

static void
applier_on_state_f(struct trigger *trigger, void *event)
{
        (void) event;
        struct applier_on_state *on_state =
                container_of(trigger, struct applier_on_state, base);

        struct applier *applier = on_state->applier;

        if (applier->state != APPLIER_OFF &&
            applier->state != APPLIER_STOPPED &&
            applier->state != on_state->desired_state)
                return;

        /* Wake up waiter */
        fiber_cond_signal(&on_state->wakeup);

        applier_pause(applier);
}

static inline void
applier_add_on_state(struct applier *applier,
                     struct applier_on_state *trigger,
                     enum applier_state desired_state)
{
        trigger_create(&trigger->base, applier_on_state_f, NULL, NULL);
        trigger->applier = applier;
        fiber_cond_create(&trigger->wakeup);
        trigger->desired_state = desired_state;
        trigger_add(&applier->on_state, &trigger->base);
}

static inline void
applier_clear_on_state(struct applier_on_state *trigger)
{
        fiber_cond_destroy(&trigger->wakeup);
        trigger_clear(&trigger->base);
}

static inline int
applier_wait_for_state(struct applier_on_state *trigger, double timeout)
{
        struct applier *applier = trigger->applier;
        double deadline = ev_monotonic_now(loop()) + timeout;
        while (applier->state != APPLIER_OFF &&
               applier->state != APPLIER_STOPPED &&
               applier->state != trigger->desired_state) {
                if (fiber_cond_wait_deadline(&trigger->wakeup, deadline) != 0)
                        return -1; /* ER_TIMEOUT */
        }
        if (applier->state != trigger->desired_state) {
                assert(applier->state == APPLIER_OFF ||
                       applier->state == APPLIER_STOPPED);
                /* Re-throw the original error */
                assert(!diag_is_empty(&applier->reader->diag));
                diag_move(&applier->reader->diag, &fiber()->diag);
                return -1;
        }
        return 0;
}

void
applier_resume_to_state(struct applier *applier, enum applier_state state,
                        double timeout)
{
        struct applier_on_state trigger;
        applier_add_on_state(applier, &trigger, state);
        applier_resume(applier);
        int rc = applier_wait_for_state(&trigger, timeout);
        applier_clear_on_state(&trigger);
        if (rc != 0)
                diag_raise();
        assert(applier->state == state);
}

tarantool / tarantool / 14944

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous