rep_log.c

/*-
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 2004-2005
 *      Sleepycat Software.  All rights reserved.
 *
 * $Id: rep_log.c,v 12.26 2005/10/12 17:58:39 bostic Exp $
 */

#include "db_config.h"

#ifndef NO_SYSTEM_INCLUDES
#if TIME_WITH_SYS_TIME
#include <sys/time.h>
#include <time.h>
#else
#if HAVE_SYS_TIME_H
#include <sys/time.h>
#else
#include <time.h>
#endif
#endif

#include <stdlib.h>
#include <string.h>
#endif

#include "db_int.h"
#include "dbinc/db_page.h"
#include "dbinc/db_am.h"
#include "dbinc/log.h"

/*
 * __rep_allreq --
 *      Handle a REP_ALL_REQ message.
 *
 * PUBLIC: int __rep_allreq __P((DB_ENV *, REP_CONTROL *, int));
 */
int
__rep_allreq(dbenv, rp, eid)
        DB_ENV *dbenv;
        REP_CONTROL *rp;
        int eid;
{
        DB_LOGC *logc;
        DB_LSN oldfilelsn;
        DB_REP *db_rep;
        DBT data_dbt;
        REP *rep;
        REP_BULK bulk;
        REP_THROTTLE repth;
        uintptr_t bulkoff;
        u_int32_t bulkflags, flags, use_bulk;
        int ret, t_ret;

        ret = 0;
        db_rep = dbenv->rep_handle;
        rep = db_rep->region;

        if ((ret = __log_cursor(dbenv, &logc)) != 0)
                return (ret);
        memset(&data_dbt, 0, sizeof(data_dbt));
        /*
         * If we're doing bulk transfer, allocate a bulk buffer to put our
         * log records in.  We still need to initialize the throttle info
         * because if we encounter a log record larger than our entire bulk
         * buffer, we need to send it as a singleton and also we want to
         * support throttling with bulk.
         *
         * Use a local var so we don't need to worry if someone else turns
         * on/off bulk in the middle of our call.
         */
        use_bulk = FLD_ISSET(rep->config, REP_C_BULK);
        if (use_bulk && (ret = __rep_bulk_alloc(dbenv, &bulk, eid,
            &bulkoff, &bulkflags, REP_BULK_LOG)) != 0)
                goto err;
        memset(&repth, 0, sizeof(repth));
        REP_SYSTEM_LOCK(dbenv);
        repth.gbytes = rep->gbytes;
        repth.bytes = rep->bytes;
        oldfilelsn = repth.lsn = rp->lsn;
        repth.type = REP_LOG;
        repth.data_dbt = &data_dbt;
        REP_SYSTEM_UNLOCK(dbenv);
        flags = IS_ZERO_LSN(rp->lsn) ||
            IS_INIT_LSN(rp->lsn) ?  DB_FIRST : DB_SET;
        /*
         * We get the first item so that a client servicing requests
         * can distinguish between not having the records and reaching
         * the end of its log.  Return the DB_NOTFOUND if the client
         * cannot get the record.  Return 0 if we finish the loop and
         * sent all that we have.
         */
        ret = __log_c_get(logc, &repth.lsn, &data_dbt, flags);
        if (ret == DB_NOTFOUND) {
                if (F_ISSET(rep, REP_F_MASTER))
                        ret = 0;
                goto err;
        }
        /*
         * For singleton log records, we break when we get a REP_LOG_MORE.
         * Or if we're not using throttling, or we are using bulk, we stop
         * when we reach the end (i.e. ret != 0).
         */
        for (;
            ret == 0 && repth.type != REP_LOG_MORE;
            ret = __log_c_get(logc, &repth.lsn, &data_dbt, DB_NEXT)) {
                /*
                 * If the client is asking for all records because it doesn't
                 * have any, and our first record is not in the first log
                 * file, then the client is outdated and needs to get a
                 * VERIFY_FAIL.
                 */
                if (repth.lsn.file != 1 && flags == DB_FIRST) {
                        (void)__rep_send_message(dbenv, eid,
                            REP_VERIFY_FAIL, &repth.lsn, NULL, 0, 0);
                        break;
                }
                if (repth.lsn.file != oldfilelsn.file)
                        (void)__rep_send_message(dbenv,
                            eid, REP_NEWFILE, &oldfilelsn, NULL, 0, 0);
                /*
                 * If we are configured for bulk, try to send this as a bulk
                 * request.  If not configured, or it is too big for bulk
                 * then just send normally.
                 */
                if (use_bulk)
                        ret = __rep_bulk_message(dbenv, &bulk, &repth,
                            &repth.lsn, &data_dbt, DB_LOG_RESEND);
                if (!use_bulk || ret == DB_REP_BULKOVF)
                        ret = __rep_send_throttle(dbenv, eid, &repth, 0);
                if (ret != 0)
                        break;
                /*
                 * If we are about to change files, then we'll need the
                 * last LSN in the previous file.  Save it here.
                 */
                oldfilelsn = repth.lsn;
                oldfilelsn.offset += logc->c_len;
        }

        if (ret == DB_NOTFOUND)
                ret = 0;
        /*
         * We're done, force out whatever remains in the bulk buffer and
         * free it.
         */
        if (use_bulk && (t_ret = __rep_bulk_free(dbenv, &bulk,
            DB_LOG_RESEND)) != 0 && ret == 0)
                ret = t_ret;
err:
        if ((t_ret = __log_c_close(logc)) != 0 && ret == 0)
                ret = t_ret;
        return (ret);
}

/*
 * __rep_log --
 *      Handle a REP_LOG/REP_LOG_MORE message.
 *
 * PUBLIC: int __rep_log __P((DB_ENV *, REP_CONTROL *, DBT *,
 * PUBLIC:     time_t, DB_LSN *));
 */
int
__rep_log(dbenv, rp, rec, savetime, ret_lsnp)
        DB_ENV *dbenv;
        REP_CONTROL *rp;
        DBT *rec;
        time_t savetime;
        DB_LSN *ret_lsnp;
{
        DB_LOG *dblp;
        DB_LSN lsn;
        DB_REP *db_rep;
        LOG *lp;
        REP *rep;
        int is_dup, master, ret;

        is_dup = ret = 0;
        db_rep = dbenv->rep_handle;
        rep = db_rep->region;
        dblp = dbenv->lg_handle;
        lp = dblp->reginfo.primary;

        ret = __rep_apply(dbenv, rp, rec, ret_lsnp, &is_dup);
        switch (ret) {
        /*
         * We're in an internal backup and we've gotten
         * all the log we need to run recovery.  Do so now.
         */
        case DB_REP_LOGREADY:
                if ((ret = __log_flush(dbenv, NULL)) != 0)
                        goto out;
                if ((ret = __rep_verify_match(dbenv, &rep->last_lsn,
                    savetime)) == 0) {
                        REP_SYSTEM_LOCK(dbenv);
                        ZERO_LSN(rep->first_lsn);
                        ZERO_LSN(rep->last_lsn);
                        F_CLR(rep, REP_F_RECOVER_LOG);
                        REP_SYSTEM_UNLOCK(dbenv);
                }
                break;
        /*
         * If we get any of the "normal" returns, we only process
         * LOG_MORE if this is not a duplicate record.  If the
         * record is a duplicate we don't want to handle LOG_MORE
         * and request a multiple data stream (or trigger internal
         * initialization) since this could be a very old record
         * that no longer exists on the master.
         */
        case DB_REP_ISPERM:
        case DB_REP_NOTPERM:
        case 0:
                if (is_dup)
                        goto out;
                else
                        break;
        /*
         * Any other return (errors), we're done.
         */
        default:
                goto out;
        }
        if (rp->rectype == REP_LOG_MORE) {
                REP_SYSTEM_LOCK(dbenv);
                master = rep->master_id;
                REP_SYSTEM_UNLOCK(dbenv);
                LOG_SYSTEM_LOCK(dbenv);
                lsn = lp->lsn;
                LOG_SYSTEM_UNLOCK(dbenv);
                /*
                 * If the master_id is invalid, this means that since
                 * the last record was sent, somebody declared an
                 * election and we may not have a master to request
                 * things of.
                 *
                 * This is not an error;  when we find a new master,
                 * we'll re-negotiate where the end of the log is and
                 * try to bring ourselves up to date again anyway.
                 *
                 * If we've asked for a bunch of records, it could
                 * either be from a LOG_REQ or ALL_REQ.  If we're
                 * waiting for a gap to be filled, call loggap_req,
                 * otherwise use ALL_REQ again.
                 */
                MUTEX_LOCK(dbenv, rep->mtx_clientdb);
                if (master == DB_EID_INVALID) {
                        ret = 0;
                        MUTEX_UNLOCK(dbenv, rep->mtx_clientdb);
                } else if (IS_ZERO_LSN(lp->waiting_lsn)) {
                        /*
                         * We're making an ALL_REQ.  However, since we're
                         * in a LOG_MORE, this is in reply to a request and
                         * it is likely we may receive new records, even if
                         * we don't have any at this moment.  So, to avoid
                         * multiple data streams, set the wait_recs high
                         * now to give the master a chance to start sending
                         * us these records before the gap code re-requests
                         * the same gap.  Wait_recs will get reset once we
                         * start receiving these records.
                         */
                        lp->wait_recs = rep->max_gap;
                        MUTEX_UNLOCK(dbenv, rep->mtx_clientdb);
                        if (__rep_send_message(dbenv, master, REP_ALL_REQ,
                            &lsn, NULL, 0, DB_REP_ANYWHERE) != 0)
                                goto out;
                } else {
                        ret = __rep_loggap_req(dbenv, rep, &lsn, REP_GAP_FORCE);
                        MUTEX_UNLOCK(dbenv, rep->mtx_clientdb);
                }
        }
out:
        return (ret);
}

/*
 * __rep_bulk_log --
 *      Handle a REP_BULK_LOG message.
 *
 * PUBLIC: int __rep_bulk_log __P((DB_ENV *, REP_CONTROL *, DBT *,
 * PUBLIC:     time_t, DB_LSN *));
 */
int
__rep_bulk_log(dbenv, rp, rec, savetime, ret_lsnp)
        DB_ENV *dbenv;
        REP_CONTROL *rp;
        DBT *rec;
        time_t savetime;
        DB_LSN *ret_lsnp;
{
        DB_REP *db_rep;
        REP *rep;
        int ret;

        db_rep = dbenv->rep_handle;
        rep = db_rep->region;

        ret = __log_rep_split(dbenv, rp, rec, ret_lsnp);
        switch (ret) {
        /*
         * We're in an internal backup and we've gotten
         * all the log we need to run recovery.  Do so now.
         */
        case DB_REP_LOGREADY:
                if ((ret = __log_flush(dbenv, NULL)) != 0)
                        goto out;
                if ((ret = __rep_verify_match(dbenv, &rep->last_lsn,
                    savetime)) == 0) {
                        REP_SYSTEM_LOCK(dbenv);
                        ZERO_LSN(rep->first_lsn);
                        ZERO_LSN(rep->last_lsn);
                        F_CLR(rep, REP_F_RECOVER_LOG);
                        REP_SYSTEM_UNLOCK(dbenv);
                }
                break;
        /*
         * Any other return (errors), we're done.
         */
        default:
                break;
        }
out:
        return (ret);
}

/*
 * __rep_log_req --
 *      Handle a REP_LOG_REQ message.
 *
 * PUBLIC: int __rep_logreq __P((DB_ENV *, REP_CONTROL *, DBT *, int));
 */
int
__rep_logreq(dbenv, rp, rec, eid)
        DB_ENV *dbenv;
        REP_CONTROL *rp;
        DBT *rec;
        int eid;
{
        DB_LOG *dblp;
        DB_LOGC *logc;
        DB_LSN endlsn, lsn, oldfilelsn;
        DB_REP *db_rep;
        DBT data_dbt;
        LOG *lp;
        REP *rep;
        REP_BULK bulk;
        REP_THROTTLE repth;
        uintptr_t bulkoff;
        u_int32_t bulkflags, use_bulk;
        int ret, t_ret;
#ifdef DIAGNOSTIC
        DB_MSGBUF mb;
#endif

        ret = 0;
        db_rep = dbenv->rep_handle;
        rep = db_rep->region;
        dblp = dbenv->lg_handle;
        lp = dblp->reginfo.primary;

        if (rec != NULL && rec->size != 0) {
                RPRINT(dbenv, rep, (dbenv, &mb,
                    "[%lu][%lu]: LOG_REQ max lsn: [%lu][%lu]",
                    (u_long) rp->lsn.file, (u_long)rp->lsn.offset,
                    (u_long)((DB_LSN *)rec->data)->file,
                    (u_long)((DB_LSN *)rec->data)->offset));
        }
        /*
         * There are three different cases here.
         * 1. We asked log_c_get for a particular LSN and got it.
         * 2. We asked log_c_get for an LSN and it's not found because it is
         *      beyond the end of a log file and we need a NEWFILE msg.
         *      and then the record that was requested.
         * 3. We asked log_c_get for an LSN and it simply doesn't exist, but
         *    doesn't meet any of those other criteria, in which case
         *    it's an error (that should never happen on a master).
         *
         * If we have a valid LSN and the request has a data_dbt with
         * it, the sender is asking for a chunk of log records.
         * Then we need to send all records up to the LSN in the data dbt.
         */
        memset(&data_dbt, 0, sizeof(data_dbt));
        oldfilelsn = lsn = rp->lsn;
        if ((ret = __log_cursor(dbenv, &logc)) != 0)
                return (ret);
        ret = __log_c_get(logc, &lsn, &data_dbt, DB_SET);

        if (ret == 0) /* Case 1 */
                (void)__rep_send_message(dbenv,
                   eid, REP_LOG, &lsn, &data_dbt, DB_LOG_RESEND, 0);
        else if (ret == DB_NOTFOUND) {
                LOG_SYSTEM_LOCK(dbenv);
                endlsn = lp->lsn;
                LOG_SYSTEM_UNLOCK(dbenv);
                if (endlsn.file > lsn.file) {
                        /*
                         * Case 2:
                         * Need to find the LSN of the last record in
                         * file lsn.file so that we can send it with
                         * the NEWFILE call.  In order to do that, we
                         * need to try to get {lsn.file + 1, 0} and
                         * then backup.
                         */
                        endlsn.file = lsn.file + 1;
                        endlsn.offset = 0;
                        if ((ret = __log_c_get(logc,
                            &endlsn, &data_dbt, DB_SET)) != 0 ||
                            (ret = __log_c_get(logc,
                                &endlsn, &data_dbt, DB_PREV)) != 0) {
                                RPRINT(dbenv, rep, (dbenv, &mb,
                                    "Unable to get prev of [%lu][%lu]",
                                    (u_long)lsn.file,
                                    (u_long)lsn.offset));
                                /*
                                 * We want to push the error back
                                 * to the client so that the client
                                 * does an internal backup.  The
                                 * client asked for a log record
                                 * we no longer have and it is
                                 * outdated.
                                 * XXX - This could be optimized by
                                 * having the master perform and
                                 * send a REP_UPDATE message.  We
                                 * currently want the client to set
                                 * up its 'update' state prior to
                                 * requesting REP_UPDATE_REQ.
                                 *
                                 * If we're a client servicing a request
                                 * just return DB_NOTFOUND.
                                 */
                                if (F_ISSET(rep, REP_F_MASTER)) {
                                        ret = 0;
                                        (void)__rep_send_message(dbenv, eid,
                                            REP_VERIFY_FAIL, &rp->lsn,
                                            NULL, 0, 0);
                                } else
                                        ret = DB_NOTFOUND;
                        } else {
                                endlsn.offset += logc->c_len;
                                (void)__rep_send_message(dbenv, eid,
                                    REP_NEWFILE, &endlsn, NULL, 0, 0);
                        }
                } else {
                        /* Case 3 */
                        /*
                         * If we're a master, this is a problem.
                         * If we're a client servicing a request
                         * just return the DB_NOTFOUND.
                         */
                        if (F_ISSET(rep, REP_F_MASTER)) {
                                __db_err(dbenv,
                                    "Request for LSN [%lu][%lu] fails",
                                    (u_long)lsn.file, (u_long)lsn.offset);
                                DB_ASSERT(0);
                                ret = EINVAL;
                        }
                }
        }
        if (ret != 0)
                goto err;

        /*
         * If the user requested a gap, send the whole thing,
         * while observing the limits from set_rep_limit.
         */
        /*
         * If we're doing bulk transfer, allocate a bulk buffer to put our
         * log records in.  We still need to initialize the throttle info
         * because if we encounter a log record larger than our entire bulk
         * buffer, we need to send it as a singleton.
         *
         * Use a local var so we don't need to worry if someone else turns
         * on/off bulk in the middle of our call.
         */
        use_bulk = FLD_ISSET(rep->config, REP_C_BULK);
        if (use_bulk && (ret = __rep_bulk_alloc(dbenv, &bulk, eid,
            &bulkoff, &bulkflags, REP_BULK_LOG)) != 0)
                goto err;
        memset(&repth, 0, sizeof(repth));
        REP_SYSTEM_LOCK(dbenv);
        repth.gbytes = rep->gbytes;
        repth.bytes = rep->bytes;
        repth.type = REP_LOG;
        repth.data_dbt = &data_dbt;
        REP_SYSTEM_UNLOCK(dbenv);
        while (ret == 0 && rec != NULL && rec->size != 0 &&
            repth.type == REP_LOG) {
                if ((ret =
                    __log_c_get(logc, &repth.lsn, &data_dbt, DB_NEXT)) != 0) {
                        /*
                         * If we're a client and we only have part of the gap,
                         * return DB_NOTFOUND so that we send a REREQUEST
                         * back to the requester and it can ask for more.
                         */
                        if (ret == DB_NOTFOUND && F_ISSET(rep, REP_F_MASTER))
                                ret = 0;
                        break;
                }
                if (log_compare(&repth.lsn, (DB_LSN *)rec->data) >= 0)
                        break;
                if (repth.lsn.file != oldfilelsn.file)
                        (void)__rep_send_message(dbenv,
                            eid, REP_NEWFILE, &oldfilelsn, NULL, 0, 0);
                /*
                 * If we are configured for bulk, try to send this as a bulk
                 * request.  If not configured, or it is too big for bulk
                 * then just send normally.
                 */
                if (use_bulk)
                        ret = __rep_bulk_message(dbenv, &bulk, &repth,
                            &repth.lsn, &data_dbt, DB_LOG_RESEND);
                if (!use_bulk || ret == DB_REP_BULKOVF)
                        ret = __rep_send_throttle(dbenv, eid, &repth, 0);
                if (ret != 0)
                        break;
                /*
                 * If we are about to change files, then we'll need the
                 * last LSN in the previous file.  Save it here.
                 */
                oldfilelsn = repth.lsn;
                oldfilelsn.offset += logc->c_len;
        }

        /*
         * We're done, force out whatever remains in the bulk buffer and
         * free it.
         */
        if (use_bulk && (t_ret = __rep_bulk_free(dbenv, &bulk,
            DB_LOG_RESEND)) != 0 && ret == 0)
                ret = t_ret;
err:
        if ((t_ret = __log_c_close(logc)) != 0 && ret == 0)
                ret = t_ret;
        return (ret);
}

/*
 * __rep_loggap_req -
 *      Request a log gap.  Assumes the caller holds the REP->mtx_clientdb.
 *
 * lsnp is the current LSN we're handling.  It is used to help decide
 *      if we ask for a gap or singleton.
 * gapflags are flags that may override the algorithm or control the
 *      processing in some way.
 *
 * PUBLIC: int __rep_loggap_req __P((DB_ENV *, REP *, DB_LSN *, u_int32_t));
 */
int
__rep_loggap_req(dbenv, rep, lsnp, gapflags)
        DB_ENV *dbenv;
        REP *rep;
        DB_LSN *lsnp;
        u_int32_t gapflags;
{
        DB_LOG *dblp;
        DBT max_lsn_dbt, *max_lsn_dbtp;
        DB_LSN next_lsn;
        LOG *lp;
        u_int32_t flags, type;

        dblp = dbenv->lg_handle;
        lp = dblp->reginfo.primary;
        LOG_SYSTEM_LOCK(dbenv);
        next_lsn = lp->lsn;
        LOG_SYSTEM_UNLOCK(dbenv);
        flags = 0;
        type = REP_LOG_REQ;

        /*
         * Check if we need to ask for the gap.
         * We ask for the gap if:
         *      We are forced to with gapflags.
         *      If max_wait_lsn is ZERO_LSN - we've never asked for
         *        records before.
         *      If we asked for a single record and received it.
         *
         * If we want a gap, but don't have an ending LSN (waiting_lsn)
         * send an ALL_REQ.  This is primarily used by REP_REREQUEST when
         * an ALL_REQ was not able to be fulfilled by another client.
         */
        if (FLD_ISSET(gapflags, (REP_GAP_FORCE | REP_GAP_REREQUEST)) ||
            IS_ZERO_LSN(lp->max_wait_lsn) ||
            (lsnp != NULL && log_compare(lsnp, &lp->max_wait_lsn) == 0)) {
                lp->max_wait_lsn = lp->waiting_lsn;
                if (IS_ZERO_LSN(lp->max_wait_lsn))
                        type = REP_ALL_REQ;
                memset(&max_lsn_dbt, 0, sizeof(max_lsn_dbt));
                max_lsn_dbt.data = &lp->waiting_lsn;
                max_lsn_dbt.size = sizeof(lp->waiting_lsn);
                max_lsn_dbtp = &max_lsn_dbt;
                /*
                 * Gap requests are "new" and can go anywhere, unless
                 * this is already a rerequest.
                 */
                if (FLD_ISSET(gapflags, REP_GAP_REREQUEST))
                        flags = DB_REP_REREQUEST;
                else
                        flags = DB_REP_ANYWHERE;
        } else {
                max_lsn_dbtp = NULL;
                lp->max_wait_lsn = next_lsn;
                /*
                 * If we're dropping to singletons, this is a rerequest.
                 */
                flags = DB_REP_REREQUEST;
        }
        if (rep->master_id != DB_EID_INVALID) {
                rep->stat.st_log_requested++;
                (void)__rep_send_message(dbenv, rep->master_id,
                    type, &next_lsn, max_lsn_dbtp, 0, flags);
        } else
                (void)__rep_send_message(dbenv, DB_EID_BROADCAST,
                    REP_MASTER_REQ, NULL, NULL, 0, 0);

        return (0);
}

---