10768987082

Committed 09 Sep 2024 07:55AM UTC coverage: 87.229% (+0.03%) from 87.204%

Build # 10768987082

Build Type

push

github

Committed by

locker

Commit Message

test: add test for #10148

The fix itself is in the small submodule which is bumped in the previous
commit.

Closes #10148

NO_DOC=bugfix

(cherry picked from commit e4ce9e111)

Run Details

68351 of 121607 branches covered (56.21%)

101036 of 115828 relevant lines covered (87.23%)

2466163.47 hits per line

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

92.45

/src/box/recovery.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 "recovery.h"

#include "small/rlist.h"
#include "scoped_guard.h"
#include "trigger.h"
#include "fiber.h"
#include "xlog.h"
#include "xrow.h"
#include "xstream.h"
#include "wal.h" /* wal_watcher */
#include "replication.h"
#include "session.h"
#include "coio_file.h"
#include "error.h"
#include "iproto_constants.h"

/*
 * Recovery subsystem
 * ------------------
 *
 * A facade of the recovery subsystem is struct recovery.
 *
 * Depending on the actual task being performed the recovery
 * can be in a different state.
 *
 * Let's enumerate all possible distinct states of recovery:
 *
 * IR - initial recovery, initiated right after server start:
 * reading data from a checkpoint and existing WALs
 * and restoring the in-memory state
 * IRR - initial replication relay mode, reading data from
 * existing WALs (xlogs) and sending it to the client.
 *
 * HS - standby mode, entered once all existing WALs are read:
 * following the WAL directory for all changes done by the master
 * and updating the in-memory state
 * RR - replication relay, following the WAL directory for all
 * changes done by the master and sending them to the
 * replica
 *
 * The following state transitions are possible/supported:
 *
 * recovery_init() -> IR | IRR # recover()
 * IR -> HS         # recovery_follow_local()
 * IRR -> RR        # recovery_follow_local()
 */

/* {{{ Initial recovery */

/**
 * Throws an exception in  case of error.
 */
struct recovery *
recovery_new(const char *wal_dirname, bool force_recovery,
             const struct vclock *vclock)
{
        struct recovery *r = (struct recovery *)
                        calloc(1, sizeof(*r));

        if (r == NULL) {
                tnt_raise(OutOfMemory, sizeof(*r), "malloc",
                          "struct recovery");
        }

        auto guard = make_scoped_guard([=]{
                free(r);
        });

        xdir_create(&r->wal_dir, wal_dirname, XLOG, &INSTANCE_UUID,
                    &xlog_opts_default);
        r->wal_dir.force_recovery = force_recovery;

        vclock_copy(&r->vclock, vclock);

        /**
         * Avoid scanning WAL dir before we recovered
         * the snapshot and know instance UUID - this will
         * make sure the scan skips files with wrong
         * UUID, see replication/cluster.test for
         * details.
         */
        xdir_check_xc(&r->wal_dir);

        r->watcher = NULL;
        rlist_create(&r->on_close_log);

        guard.is_active = false;
        return r;
}

void
recovery_scan(struct recovery *r, struct vclock *end_vclock,
              struct vclock *gc_vclock, struct xstream *stream)
{
        xdir_scan_xc(&r->wal_dir, true);

        if (xdir_last_vclock(&r->wal_dir, end_vclock) < 0 ||
            vclock_compare(end_vclock, &r->vclock) < 0) {
                /* No xlogs after last checkpoint. */
                vclock_copy(gc_vclock, &r->vclock);
                vclock_copy(end_vclock, &r->vclock);
                return;
        }

        if (xdir_first_vclock(&r->wal_dir, gc_vclock) < 0)
                unreachable();

        /* Scan the last xlog to find end vclock. */
        struct xlog_cursor cursor;
        if (xdir_open_cursor(&r->wal_dir, vclock_sum(end_vclock), &cursor) != 0)
                return;
        struct xrow_header row;
        while (xlog_cursor_next(&cursor, &row, true) == 0) {
                vclock_follow_xrow(end_vclock, &row);
                if (++stream->row_count % WAL_ROWS_PER_YIELD == 0)
                        xstream_yield(stream);
        }
        xlog_cursor_close(&cursor, false);

        xstream_reset(stream);
}

static inline void
recovery_close_log(struct recovery *r)
{
        if (!xlog_cursor_is_open(&r->cursor))
                return;
        if (xlog_cursor_is_eof(&r->cursor)) {
                say_info("done `%s'", r->cursor.name);
        } else {
                say_warn("file `%s` wasn't correctly closed",
                         r->cursor.name);
        }
        xlog_cursor_close(&r->cursor, false);
        trigger_run_xc(&r->on_close_log, NULL);
}

static void
recovery_open_log(struct recovery *r, const struct vclock *vclock)
{
        XlogGapError *e;
        struct xlog_meta meta = r->cursor.meta;
        enum xlog_cursor_state state = r->cursor.state;

        recovery_close_log(r);

        xdir_open_cursor_xc(&r->wal_dir, vclock_sum(vclock), &r->cursor);

        if (state == XLOG_CURSOR_NEW &&
            vclock_compare(vclock, &r->vclock) > 0) {
                /*
                 * This is the first WAL we are about to scan
                 * and the best clock we could find is greater
                 * or is incomparable with the initial recovery
                 * position.
                 */
                goto gap_error;
        }

        if (state != XLOG_CURSOR_NEW &&
            vclock_is_set(&r->cursor.meta.prev_vclock) &&
            vclock_compare(&r->cursor.meta.prev_vclock, &meta.vclock) != 0) {
                /*
                 * WALs are missing between the last scanned WAL
                 * and the next one.
                 */
                goto gap_error;
        }
out:
        /*
         * We must promote recovery clock even if we don't recover
         * anything from the next WAL. Otherwise if the last WAL
         * in the directory is corrupted or empty and the previous
         * one has an LSN gap at the end (due to a write error),
         * we will create the next WAL between two existing ones,
         * thus breaking the file order.
         */
        if (vclock_compare(&r->vclock, vclock) < 0)
                vclock_copy(&r->vclock, vclock);
        return;

gap_error:
        e = tnt_error(XlogGapError, &r->vclock, vclock);
        if (!r->wal_dir.force_recovery)
                throw e;
        /* Ignore missing WALs if force_recovery is set. */
        e->log();
        say_warn("ignoring a gap in LSN");
        goto out;
}

void
recovery_delete(struct recovery *r)
{
        recovery_stop_local(r);

        trigger_destroy(&r->on_close_log);
        xdir_destroy(&r->wal_dir);
        if (xlog_cursor_is_open(&r->cursor)) {
                /*
                 * Possible if shutting down a replication
                 * relay or if error during startup.
                 */
                xlog_cursor_close(&r->cursor, false);
        }
        free(r);
}

/**
 * Read all rows in a file starting from the last position.
 * Advance the position. If end of file is reached,
 * set l.eof_read.
 * The reading will be stopped on reaching stop_vclock.
 * Use NULL for boundless recover
 */
static void
recover_xlog(struct recovery *r, struct xstream *stream,
             const struct vclock *stop_vclock)
{
        struct xrow_header row;
        bool is_sending_tx = false;
        while (xlog_cursor_next_xc(&r->cursor, &row,
                                   r->wal_dir.force_recovery) == 0) {
                if (++stream->row_count % WAL_ROWS_PER_YIELD == 0) {
                        xstream_yield(stream);
                }
                if (stream->row_count % 100000 == 0) {
                        say_info_ratelimited("%.1fM rows processed",
                                             stream->row_count / 1e6);
                }
                /*
                 * Read the next row from xlog file.
                 *
                 * xlog_cursor_next_xc() returns 1 when
                 * it can not read more rows. This doesn't mean
                 * the file is fully read: it's fully read only
                 * when EOF marker has been read, see i.eof_read
                 */
                if (stop_vclock != NULL &&
                    r->vclock.signature >= stop_vclock->signature)
                        return;

                /*
                 * All rows in xlog files have an assigned replica
                 * id. The only exception are local rows, which
                 * are signed with a zero replica id.
                 */
                assert(row.replica_id != 0 || row.group_id == GROUP_LOCAL);
                int64_t current_lsn = vclock_get(&r->vclock, row.replica_id);
                if (row.lsn <= current_lsn) {
                        /*
                         * Skip the already applied row, if it is not needed to
                         * preserve transaction boundaries (is not the last row
                         * of a currently recovered transaction). Otherwise,
                         * replace it with a NOP, so that the transaction end
                         * flag reaches the receiver, but the data isn't
                         * recovered twice.
                         */
                        if (!is_sending_tx || !row.is_commit)
                                continue; /* already applied, skip */
                        row.type = IPROTO_NOP;
                        row.bodycnt = 0;
                        row.body[0].iov_base = NULL;
                        row.body[0].iov_len = 0;
                } else {
                        /*
                         * We can promote the vclock either before or
                         * after xstream_write(): it only makes any impact
                         * in case of forced recovery, when we skip the
                         * failed row anyway.
                         */
                        vclock_follow_xrow(&r->vclock, &row);
                }
                is_sending_tx = !row.is_commit;
                if (xstream_write(stream, &row) != 0) {
                        if (!r->wal_dir.force_recovery)
                                diag_raise();

                        say_error("skipping row {%u: %lld}",
                                  (unsigned)row.replica_id, (long long)row.lsn);
                        diag_log();
                }
        }
}

/**
 * Find out if there are new .xlog files since the current
 * LSN, and read them all up.
 *
 * Reading will be stopped on reaching recovery
 * vclock signature > to_checkpoint (after playing to_checkpoint record)
 * use NULL for boundless recover
 *
 * This function will not close r->current_wal if
 * recovery was successful.
 */
void
recover_remaining_wals(struct recovery *r, struct xstream *stream,
                       const struct vclock *stop_vclock, bool scan_dir)
{
        struct vclock *clock;

        if (scan_dir)
                xdir_scan_xc(&r->wal_dir, true);

        if (xlog_cursor_is_open(&r->cursor)) {
                /* If there's a WAL open, recover from it first. */
                assert(!xlog_cursor_is_eof(&r->cursor));
                clock = vclockset_search(&r->wal_dir.index,
                                         &r->cursor.meta.vclock);
                if (clock != NULL)
                        goto recover_current_wal;
                /*
                 * The current WAL has disappeared under our feet -
                 * assume anything can happen in production and go on.
                 */
                say_error("file `%s' was deleted under our feet",
                          r->cursor.name);
        }

        for (clock = vclockset_match(&r->wal_dir.index, &r->vclock);
             clock != NULL;
             clock = vclockset_next(&r->wal_dir.index, clock)) {
                if (stop_vclock != NULL &&
                    clock->signature >= stop_vclock->signature) {
                        break;
                }

                if (xlog_cursor_is_eof(&r->cursor) &&
                    vclock_sum(&r->cursor.meta.vclock) >= vclock_sum(clock)) {
                        /*
                         * If we reached EOF while reading last xlog,
                         * we don't need to rescan it.
                         */
                        continue;
                }

                recovery_open_log(r, clock);

                say_info("recover from `%s'", r->cursor.name);

recover_current_wal:
                recover_xlog(r, stream, stop_vclock);
        }

        if (xlog_cursor_is_eof(&r->cursor))
                recovery_close_log(r);

        if (stop_vclock != NULL && vclock_compare(&r->vclock, stop_vclock) != 0)
                tnt_raise(XlogGapError, &r->vclock, stop_vclock);
}

void
recovery_finalize(struct recovery *r)
{
        recovery_close_log(r);
}


/* }}} */

/* {{{ Local recovery: support of hot standby and replication relay */

/**
 * Implements a subscription to WAL updates via fs events.
 * Any change to the WAL dir itself or a change in the XLOG
 * file triggers a wakeup. The WAL dir path is set in the
 * constructor. XLOG file path is set with set_log_path().
 */
class WalSubscription {
public:
        struct fiber *f;
        unsigned events;
        struct ev_stat dir_stat;
        struct ev_stat file_stat;
        char dir_path[PATH_MAX];
        char file_path[PATH_MAX];

        static void dir_stat_cb(struct ev_loop *, struct ev_stat *stat, int)
        {
                ((WalSubscription *)stat->data)->wakeup(WAL_EVENT_ROTATE);
        }

        static void file_stat_cb(struct ev_loop *, struct ev_stat *stat, int)
        {
                ((WalSubscription *)stat->data)->wakeup(WAL_EVENT_WRITE);
        }

        void wakeup(unsigned events)
        {
                this->events |= events;
                fiber_wakeup(f);
        }

        WalSubscription(const char *wal_dir)
        {
                f = fiber();
                events = 0;
                if ((size_t)snprintf(dir_path, sizeof(dir_path), "%s", wal_dir) >=
                                sizeof(dir_path)) {

                        panic("path too long: %s", wal_dir);
                }

                ev_stat_init(&dir_stat, dir_stat_cb, "", 0.0);
                ev_stat_init(&file_stat, file_stat_cb, "", 0.0);
                dir_stat.data = this;
                file_stat.data = this;

                ev_stat_set(&dir_stat, dir_path, 0.0);
                ev_stat_start(loop(), &dir_stat);
        }

        ~WalSubscription()
        {
                ev_stat_stop(loop(), &file_stat);
                ev_stat_stop(loop(), &dir_stat);
        }

        void set_log_path(const char *path)
        {
                /*
                 * Avoid toggling ev_stat if the path didn't change.
                 * Note: .file_path valid iff file_stat is active.
                 */
                if (path && ev_is_active(&file_stat) &&
                                strcmp(file_path, path) == 0) {

                        return;
                }

                ev_stat_stop(loop(), &file_stat);

                if (path == NULL)
                        return;

                if ((size_t)snprintf(file_path, sizeof(file_path), "%s", path) >=
                                sizeof(file_path)) {

                        panic("path too long: %s", path);
                }
                ev_stat_set(&file_stat, file_path, 0.0);
                ev_stat_start(loop(), &file_stat);
        }
};

static int
hot_standby_f(va_list ap)
{
        struct recovery *r = va_arg(ap, struct recovery *);
        struct xstream *stream = va_arg(ap, struct xstream *);
        bool scan_dir = true;

        ev_tstamp wal_dir_rescan_delay = va_arg(ap, ev_tstamp);
        fiber_set_user(fiber(), &admin_credentials);

        WalSubscription subscription(r->wal_dir.dirname);

        while (! fiber_is_cancelled()) {
                FiberGCChecker gc_check;
                /*
                 * Recover until there is no new stuff which appeared in
                 * the log dir while recovery was running.
                 *
                 * Use vclock signature to represent the current wal
                 * since the xlog object itself may be freed in
                 * recover_remaining_rows().
                 */
                int64_t start, end;
                do {
                        start = vclock_sum(&r->vclock);
                        try {
                                recover_remaining_wals(r, stream, NULL,
                                                       scan_dir);
                        } catch (Exception *e) {
                                e->log();
                        }
                        end = vclock_sum(&r->vclock);
                        /*
                         * Continue, given there's been progress *and* there is a
                         * chance new WALs have appeared since.
                         * Sic: end * is < start (is 0) if someone deleted all logs
                         * on the filesystem.
                         */
                } while (end > start && !xlog_cursor_is_open(&r->cursor));

                subscription.set_log_path(xlog_cursor_is_open(&r->cursor) ?
                                          r->cursor.name : NULL);

                bool timed_out = false;
                ev_tstamp deadline = ev_monotonic_now(loop())
                                     + wal_dir_rescan_delay;
                while (r->watcher != NULL &&
                       !timed_out &&
                       subscription.events == 0) {
                        timed_out = fiber_yield_deadline(deadline);
                }

                scan_dir = timed_out ||
                        (subscription.events & WAL_EVENT_ROTATE) != 0;

                subscription.events = 0;
        }
        return 0;
}

void
recovery_follow_local(struct recovery *r, struct xstream *stream,
                      const char *name, ev_tstamp wal_dir_rescan_delay)
{
        /*
         * Start 'hot_standby' background fiber to follow xlog changes.
         * It will pick up from the position of the currently open
         * xlog.
         */
        assert(r->watcher == NULL);
        r->watcher = fiber_new_system(name, hot_standby_f);
        if (r->watcher == NULL)
                diag_raise();
        fiber_set_joinable(r->watcher, true);
        fiber_start(r->watcher, r, stream, wal_dir_rescan_delay);
}

void
recovery_stop_local(struct recovery *r)
{
        if (r->watcher) {
                struct fiber *f = r->watcher;
                /* Break the loop in hot_standby_f(). */
                r->watcher = NULL;
                fiber_cancel(f);
                if (fiber_join(f) != 0)
                        diag_raise();
        }
}

/* }}} */

1	/*
2	* Copyright 2010-2016, Tarantool AUTHORS, please see AUTHORS file.
3	*
4	* Redistribution and use in source and binary forms, with or
5	* without modification, are permitted provided that the following
6	* conditions are met:
7	*
8	* 1. Redistributions of source code must retain the above
9	* copyright notice, this list of conditions and the
10	* following disclaimer.
11	*
12	* 2. Redistributions in binary form must reproduce the above
13	* copyright notice, this list of conditions and the following
14	* disclaimer in the documentation and/or other materials
15	* provided with the distribution.
16	*
17	* THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
18	* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
19	* TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
20	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
21	* <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
22	* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23	* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
24	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
25	* BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
26	* LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
28	* THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
29	* SUCH DAMAGE.
30	*/
31	#include "recovery.h"
32
33	#include "small/rlist.h"
34	#include "scoped_guard.h"
35	#include "trigger.h"
36	#include "fiber.h"
37	#include "xlog.h"
38	#include "xrow.h"
39	#include "xstream.h"
40	#include "wal.h" /* wal_watcher */
41	#include "replication.h"
42	#include "session.h"
43	#include "coio_file.h"
44	#include "error.h"
45	#include "iproto_constants.h"
46
47	/*
48	* Recovery subsystem
49	* ------------------
50	*
51	* A facade of the recovery subsystem is struct recovery.
52	*
53	* Depending on the actual task being performed the recovery
54	* can be in a different state.
55	*
56	* Let's enumerate all possible distinct states of recovery:
57	*
58	* IR - initial recovery, initiated right after server start:
59	* reading data from a checkpoint and existing WALs
60	* and restoring the in-memory state
61	* IRR - initial replication relay mode, reading data from
62	* existing WALs (xlogs) and sending it to the client.
63	*
64	* HS - standby mode, entered once all existing WALs are read:
65	* following the WAL directory for all changes done by the master
66	* and updating the in-memory state
67	* RR - replication relay, following the WAL directory for all
68	* changes done by the master and sending them to the
69	* replica
70	*
71	* The following state transitions are possible/supported:
72	*
73	* recovery_init() -> IR \| IRR # recover()
74	* IR -> HS # recovery_follow_local()
75	* IRR -> RR # recovery_follow_local()
76	*/
77
78	/* {{{ Initial recovery */
79
80	/**
81	* Throws an exception in case of error.
82	*/
83	struct recovery *
84	recovery_new(const char *wal_dirname, bool force_recovery,	3,320✔
85	const struct vclock *vclock)
86	{
87	struct recovery r = (struct recovery )
88	calloc(1, sizeof(*r));	3,320✔
89
90	if (r == NULL) {	3,320!
91	tnt_raise(OutOfMemory, sizeof(*r), "malloc",	×
92	"struct recovery");
93	}
94
95	auto guard = make_scoped_guard([=]{	×
96	free(r);	×
97	});	3,320!
98
99	xdir_create(&r->wal_dir, wal_dirname, XLOG, &INSTANCE_UUID,	3,320!
100	&xlog_opts_default);
101	r->wal_dir.force_recovery = force_recovery;	3,320✔
102
103	vclock_copy(&r->vclock, vclock);	3,320✔
104
105	/**
106	* Avoid scanning WAL dir before we recovered
107	* the snapshot and know instance UUID - this will
108	* make sure the scan skips files with wrong
109	* UUID, see replication/cluster.test for
110	* details.
111	*/
112	xdir_check_xc(&r->wal_dir);	3,320!
113
114	r->watcher = NULL;	3,320✔
115	rlist_create(&r->on_close_log);	3,320✔
116
117	guard.is_active = false;	3,320✔
118	return r;	6,640✔
119	}
120
121	void
122	recovery_scan(struct recovery r, struct vclock end_vclock,	778✔
123	struct vclock gc_vclock, struct xstream stream)
124	{
125	xdir_scan_xc(&r->wal_dir, true);	778!
126
127	if (xdir_last_vclock(&r->wal_dir, end_vclock) < 0 \|\|	1,350!
128	vclock_compare(end_vclock, &r->vclock) < 0) {	572✔
129	/* No xlogs after last checkpoint. */
130	vclock_copy(gc_vclock, &r->vclock);	210✔
131	vclock_copy(end_vclock, &r->vclock);	210✔
132	return;	210✔
133	}
134
135	if (xdir_first_vclock(&r->wal_dir, gc_vclock) < 0)	568!
136	unreachable();	×
137
138	/* Scan the last xlog to find end vclock. */
139	struct xlog_cursor cursor;
140	if (xdir_open_cursor(&r->wal_dir, vclock_sum(end_vclock), &cursor) != 0)	568!
141	return;	×
142	struct xrow_header row;
143	while (xlog_cursor_next(&cursor, &row, true) == 0) {	864!
144	vclock_follow_xrow(end_vclock, &row);	297!
145	if (++stream->row_count % WAL_ROWS_PER_YIELD == 0)	296!
146	xstream_yield(stream);	×
147	}
148	xlog_cursor_close(&cursor, false);	567!
149
150	xstream_reset(stream);	567!
151	}
152
153	static inline void
154	recovery_close_log(struct recovery *r)	5,057✔
155	{
156	if (!xlog_cursor_is_open(&r->cursor))	5,057✔
157	return;	3,705✔
158	if (xlog_cursor_is_eof(&r->cursor)) {	1,352✔
159	say_info("done `%s'", r->cursor.name);	1,330✔
160	} else {
161	say_warn("file `%s` wasn't correctly closed",	22!
162	r->cursor.name);
163	}
164	xlog_cursor_close(&r->cursor, false);	1,352✔
165	trigger_run_xc(&r->on_close_log, NULL);	1,352✔
166	}
167
168	static void
169	recovery_open_log(struct recovery r, const struct vclock vclock)	3,761✔
170	{
171	XlogGapError *e;
172	struct xlog_meta meta = r->cursor.meta;	3,761✔
173	enum xlog_cursor_state state = r->cursor.state;	3,761✔
174
175	recovery_close_log(r);	3,761!
176
177	xdir_open_cursor_xc(&r->wal_dir, vclock_sum(vclock), &r->cursor);	3,761!
178
179	if (state == XLOG_CURSOR_NEW &&	6,726✔
180	vclock_compare(vclock, &r->vclock) > 0) {	2,963✔
181	/*
182	* This is the first WAL we are about to scan
183	* and the best clock we could find is greater
184	* or is incomparable with the initial recovery
185	* position.
186	*/
187	goto gap_error;	24✔
188	}
189
190	if (state != XLOG_CURSOR_NEW &&	800✔
191	vclock_is_set(&r->cursor.meta.prev_vclock) &&	4,539✔
192	vclock_compare(&r->cursor.meta.prev_vclock, &meta.vclock) != 0) {	792✔
193	/*
194	* WALs are missing between the last scanned WAL
195	* and the next one.
196	*/
197	goto gap_error;	8✔
198	}
199	out:	3,731✔
200	/*
201	* We must promote recovery clock even if we don't recover
202	* anything from the next WAL. Otherwise if the last WAL
203	* in the directory is corrupted or empty and the previous
204	* one has an LSN gap at the end (due to a write error),
205	* we will create the next WAL between two existing ones,
206	* thus breaking the file order.
207	*/
208	if (vclock_compare(&r->vclock, vclock) < 0)	3,739✔
209	vclock_copy(&r->vclock, vclock);	9✔
210	return;	7,478✔
211
212	gap_error:	32✔
213	e = tnt_error(XlogGapError, &r->vclock, vclock);	32!
214	if (!r->wal_dir.force_recovery)	32✔
215	throw e;	24✔
216	/* Ignore missing WALs if force_recovery is set. */
217	e->log();	8!
218	say_warn("ignoring a gap in LSN");	8!
219	goto out;	8✔
220	}
221
222	void
223	recovery_delete(struct recovery *r)	3,315✔
224	{
225	recovery_stop_local(r);	3,315✔
226
227	trigger_destroy(&r->on_close_log);	3,314✔
228	xdir_destroy(&r->wal_dir);	3,315✔
229	if (xlog_cursor_is_open(&r->cursor)) {	3,316✔
230	/*
231	* Possible if shutting down a replication
232	* relay or if error during startup.
233	*/
234	xlog_cursor_close(&r->cursor, false);	2,410✔
235	}
236	free(r);	3,316✔
237	}	3,316✔
238
239	/**
240	* Read all rows in a file starting from the last position.
241	* Advance the position. If end of file is reached,
242	* set l.eof_read.
243	* The reading will be stopped on reaching stop_vclock.
244	* Use NULL for boundless recover
245	*/
246	static void
247	recover_xlog(struct recovery r, struct xstream stream,	21,270✔
248	const struct vclock *stop_vclock)
249	{
250	struct xrow_header row;
251	bool is_sending_tx = false;	21,270✔
252	while (xlog_cursor_next_xc(&r->cursor, &row,	383,835✔
253	r->wal_dir.force_recovery) == 0) {	405,105!
254	if (++stream->row_count % WAL_ROWS_PER_YIELD == 0) {	383,764!
255	xstream_yield(stream);	×
256	}
257	if (stream->row_count % 100000 == 0) {	383,764!
258	say_info_ratelimited("%.1fM rows processed",	×
259	stream->row_count / 1e6);
260	}
261	/*
262	* Read the next row from xlog file.
263	*
264	* xlog_cursor_next_xc() returns 1 when
265	* it can not read more rows. This doesn't mean
266	* the file is fully read: it's fully read only
267	* when EOF marker has been read, see i.eof_read
268	*/
269	if (stop_vclock != NULL &&	383,764✔
270	r->vclock.signature >= stop_vclock->signature)	29,206✔
271	return;	33✔
272
273	/*
274	* All rows in xlog files have an assigned replica
275	* id. The only exception are local rows, which
276	* are signed with a zero replica id.
277	*/
278	assert(row.replica_id != 0 \|\| row.group_id == GROUP_LOCAL);	383,731!
279	int64_t current_lsn = vclock_get(&r->vclock, row.replica_id);	383,731✔
280	if (row.lsn <= current_lsn) {	383,733✔
281	/*
282	* Skip the already applied row, if it is not needed to
283	* preserve transaction boundaries (is not the last row
284	* of a currently recovered transaction). Otherwise,
285	* replace it with a NOP, so that the transaction end
286	* flag reaches the receiver, but the data isn't
287	* recovered twice.
288	*/
289	if (!is_sending_tx \|\| !row.is_commit)	100,387!
290	continue; /* already applied, skip */	100,385✔
291	row.type = IPROTO_NOP;	2✔
292	row.bodycnt = 0;	2✔
293	row.body[0].iov_base = NULL;	2✔
294	row.body[0].iov_len = 0;	2✔
295	} else {
296	/*
297	* We can promote the vclock either before or
298	* after xstream_write(): it only makes any impact
299	* in case of forced recovery, when we skip the
300	* failed row anyway.
301	*/
302	vclock_follow_xrow(&r->vclock, &row);	283,346!
303	}
304	is_sending_tx = !row.is_commit;	283,339✔
305	if (xstream_write(stream, &row) != 0) {	283,339!
306	if (!r->wal_dir.force_recovery)	8✔
307	diag_raise();	5✔
308
309	say_error("skipping row {%u: %lld}",	3!
310	(unsigned)row.replica_id, (long long)row.lsn);
311	diag_log();	3!
312	}
313	}
314	}
315
316	/**
317	* Find out if there are new .xlog files since the current
318	* LSN, and read them all up.
319	*
320	* Reading will be stopped on reaching recovery
321	* vclock signature > to_checkpoint (after playing to_checkpoint record)
322	* use NULL for boundless recover
323	*
324	* This function will not close r->current_wal if
325	* recovery was successful.
326	*/
327	void
328	recover_remaining_wals(struct recovery r, struct xstream stream,	20,955✔
329	const struct vclock *stop_vclock, bool scan_dir)
330	{
331	struct vclock *clock;
332
333	if (scan_dir)	20,955✔
334	xdir_scan_xc(&r->wal_dir, true);	2,680✔
335
336	if (xlog_cursor_is_open(&r->cursor)) {	20,955✔
337	/* If there's a WAL open, recover from it first. */
338	assert(!xlog_cursor_is_eof(&r->cursor));	17,543!
339	clock = vclockset_search(&r->wal_dir.index,	35,076✔
340	&r->cursor.meta.vclock);	17,541✔
341	if (clock != NULL)	17,535!
342	goto recover_current_wal;	17,535✔
343	/*
344	* The current WAL has disappeared under our feet -
345	* assume anything can happen in production and go on.
346	*/
347	say_error("file `%s' was deleted under our feet",	×
348	r->cursor.name);
349	}
350
351	for (clock = vclockset_match(&r->wal_dir.index, &r->vclock);	3,412✔
352	clock != NULL;	24,703✔
353	clock = vclockset_next(&r->wal_dir.index, clock)) {	21,273✔
354	if (stop_vclock != NULL &&	3,764✔
355	clock->signature >= stop_vclock->signature) {	658✔
356	break;	1✔
357	}
358
359	if (xlog_cursor_is_eof(&r->cursor) &&	4,562✔
360	vclock_sum(&r->cursor.meta.vclock) >= vclock_sum(clock)) {	799✔
361	/*
362	* If we reached EOF while reading last xlog,
363	* we don't need to rescan it.
364	*/
365	continue;	2✔
366	}
367
368	recovery_open_log(r, clock);	3,761✔
369
370	say_info("recover from `%s'", r->cursor.name);	3,739✔
371
372	recover_current_wal:	2✔
373	recover_xlog(r, stream, stop_vclock);	21,272✔
374	}
375
376	if (xlog_cursor_is_eof(&r->cursor))	20,940✔
377	recovery_close_log(r);	552✔
378
379	if (stop_vclock != NULL && vclock_compare(&r->vclock, stop_vclock) != 0)	20,940!
380	tnt_raise(XlogGapError, &r->vclock, stop_vclock);	×
381	}	20,940✔
382
383	void
384	recovery_finalize(struct recovery *r)	744✔
385	{
386	recovery_close_log(r);	744✔
387	}	744✔
388
389
390	/* }}} */
391
392	/* {{{ Local recovery: support of hot standby and replication relay */
393
394	/**
395	* Implements a subscription to WAL updates via fs events.
396	* Any change to the WAL dir itself or a change in the XLOG
397	* file triggers a wakeup. The WAL dir path is set in the
398	* constructor. XLOG file path is set with set_log_path().
399	*/
400	class WalSubscription {
401	public:
402	struct fiber *f;
403	unsigned events;
404	struct ev_stat dir_stat;
405	struct ev_stat file_stat;
406	char dir_path[PATH_MAX];
407	char file_path[PATH_MAX];
408
409	static void dir_stat_cb(struct ev_loop , struct ev_stat stat, int)	3✔
410	{
411	((WalSubscription *)stat->data)->wakeup(WAL_EVENT_ROTATE);	3✔
412	}	3✔
413
414	static void file_stat_cb(struct ev_loop , struct ev_stat stat, int)	9✔
415	{
416	((WalSubscription *)stat->data)->wakeup(WAL_EVENT_WRITE);	9✔
417	}	9✔
418
419	void wakeup(unsigned events)	12✔
420	{
421	this->events \|= events;	12✔
422	fiber_wakeup(f);	12✔
423	}	12✔
424
425	WalSubscription(const char *wal_dir)	3✔
426	{	3✔
427	f = fiber();	3!
428	events = 0;	3✔
429	if ((size_t)snprintf(dir_path, sizeof(dir_path), "%s", wal_dir) >=	3!
430	sizeof(dir_path)) {
431
432	panic("path too long: %s", wal_dir);	×
433	}
434
435	ev_stat_init(&dir_stat, dir_stat_cb, "", 0.0);	3✔
436	ev_stat_init(&file_stat, file_stat_cb, "", 0.0);	3✔
437	dir_stat.data = this;	3✔
438	file_stat.data = this;	3✔
439
440	ev_stat_set(&dir_stat, dir_path, 0.0);	3✔
441	ev_stat_start(loop(), &dir_stat);	3!
442	}	3✔
443
444	~WalSubscription()	3✔
445	{	3✔
446	ev_stat_stop(loop(), &file_stat);	3!
447	ev_stat_stop(loop(), &dir_stat);	3!
448	}	3✔
449
450	void set_log_path(const char *path)	15✔
451	{
452	/*
453	* Avoid toggling ev_stat if the path didn't change.
454	* Note: .file_path valid iff file_stat is active.
455	*/
456	if (path && ev_is_active(&file_stat) &&	15✔
457	strcmp(file_path, path) == 0) {	11!
458
459	return;	11✔
460	}
461
462	ev_stat_stop(loop(), &file_stat);	4!
463
464	if (path == NULL)	4✔
465	return;	1✔
466
467	if ((size_t)snprintf(file_path, sizeof(file_path), "%s", path) >=	3!
468	sizeof(file_path)) {
469
470	panic("path too long: %s", path);	×
471	}
472	ev_stat_set(&file_stat, file_path, 0.0);	3✔
473	ev_stat_start(loop(), &file_stat);	3!
474	}
475	};
476
477	static int
478	hot_standby_f(va_list ap)	3✔
479	{
480	struct recovery r = va_arg(ap, struct recovery );	3✔
481	struct xstream stream = va_arg(ap, struct xstream );	3✔
482	bool scan_dir = true;	3✔
483
484	ev_tstamp wal_dir_rescan_delay = va_arg(ap, ev_tstamp);	3✔
485	fiber_set_user(fiber(), &admin_credentials);	3!
486
487	WalSubscription subscription(r->wal_dir.dirname);	3!
488
489	while (! fiber_is_cancelled()) {	18✔
490	FiberGCChecker gc_check;	×
491	/*
492	* Recover until there is no new stuff which appeared in
493	* the log dir while recovery was running.
494	*
495	* Use vclock signature to represent the current wal
496	* since the xlog object itself may be freed in
497	* recover_remaining_rows().
498	*/
499	int64_t start, end;
500	do {	×
501	start = vclock_sum(&r->vclock);	15✔
502	try {
503	recover_remaining_wals(r, stream, NULL,	15✔
504	scan_dir);
505	} catch (Exception *e) {	2✔
506	e->log();	1!
507	}
508	end = vclock_sum(&r->vclock);	15✔
509	/*
510	* Continue, given there's been progress and there is a
511	* chance new WALs have appeared since.
512	* Sic: end * is < start (is 0) if someone deleted all logs
513	* on the filesystem.
514	*/
515	} while (end > start && !xlog_cursor_is_open(&r->cursor));	15!
516
517	subscription.set_log_path(xlog_cursor_is_open(&r->cursor) ?	15!
518	r->cursor.name : NULL);
519
520	bool timed_out = false;	15✔
521	ev_tstamp deadline = ev_monotonic_now(loop())	15!
522	+ wal_dir_rescan_delay;	15✔
523	while (r->watcher != NULL &&	27!
524	!timed_out &&	30✔
525	subscription.events == 0) {	27✔
526	timed_out = fiber_yield_deadline(deadline);	15!
527	}
528
529	scan_dir = timed_out \|\|	30!
530	(subscription.events & WAL_EVENT_ROTATE) != 0;	15✔
531
532	subscription.events = 0;	15✔
533	}
534	return 0;	6✔
535	}
536
537	void
538	recovery_follow_local(struct recovery r, struct xstream stream,	3✔
539	const char *name, ev_tstamp wal_dir_rescan_delay)
540	{
541	/*
542	* Start 'hot_standby' background fiber to follow xlog changes.
543	* It will pick up from the position of the currently open
544	* xlog.
545	*/
546	assert(r->watcher == NULL);	3!
547	r->watcher = fiber_new_system(name, hot_standby_f);	3✔
548	if (r->watcher == NULL)	3!
549	diag_raise();	×
550	fiber_set_joinable(r->watcher, true);	3✔
551	fiber_start(r->watcher, r, stream, wal_dir_rescan_delay);	3✔
552	}	3✔
553
554	void
555	recovery_stop_local(struct recovery *r)	3,318✔
556	{
557	if (r->watcher) {	3,318✔
558	struct fiber *f = r->watcher;	3✔
559	/* Break the loop in hot_standby_f(). */
560	r->watcher = NULL;	3✔
561	fiber_cancel(f);	3✔
562	if (fiber_join(f) != 0)	3!
563	diag_raise();	×
564	}
565	}	3,318✔
566
567	/* }}} */

tarantool / tarantool / 10768987082

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