/*- * See the file LICENSE for redistribution information. * * Copyright (c) 2001, 2010 Oracle and/or its affiliates. All rights reserved. * * $Id$ */ #include "db_config.h" #include "db_int.h" #include "dbinc/db_page.h" #include "dbinc/btree.h" #include "dbinc/mp.h" #include "dbinc/txn.h" static int __rep_abort_prepared __P((ENV *)); static int __rep_await_condition __P((ENV *, struct rep_waitgoal *, db_timeout_t)); static int __rep_bt_cmp __P((DB *, const DBT *, const DBT *)); static int __rep_check_applied __P((ENV *, DB_THREAD_INFO *, DB_COMMIT_INFO *, struct rep_waitgoal *)); static void __rep_config_map __P((ENV *, u_int32_t *, u_int32_t *)); static u_int32_t __rep_conv_vers __P((ENV *, u_int32_t)); static int __rep_open_lsn_history __P((ENV *, DB_THREAD_INFO *, DB_TXN *, u_int32_t, DB **)); static int __rep_read_lsn_history __P((ENV *, DB_THREAD_INFO *, DB_TXN **, DBC **, u_int32_t, __rep_lsn_hist_data_args *, struct rep_waitgoal *, u_int32_t)); static int __rep_restore_prepared __P((ENV *)); static int __rep_save_lsn_hist __P((ENV *, DB_THREAD_INFO *, DB_LSN *)); /* * __rep_env_create -- * Replication-specific initialization of the ENV structure. * * PUBLIC: int __rep_env_create __P((DB_ENV *)); */ int __rep_env_create(dbenv) DB_ENV *dbenv; { DB_REP *db_rep; ENV *env; int ret; env = dbenv->env; if ((ret = __os_calloc(env, 1, sizeof(DB_REP), &db_rep)) != 0) return (ret); db_rep->eid = DB_EID_INVALID; db_rep->bytes = REP_DEFAULT_THROTTLE; DB_TIMEOUT_TO_TIMESPEC(DB_REP_REQUEST_GAP, &db_rep->request_gap); DB_TIMEOUT_TO_TIMESPEC(DB_REP_MAX_GAP, &db_rep->max_gap); db_rep->elect_timeout = 2 * US_PER_SEC; /* 2 seconds */ db_rep->chkpt_delay = 30 * US_PER_SEC; /* 30 seconds */ db_rep->my_priority = DB_REP_DEFAULT_PRIORITY; /* * Make no clock skew the default. Setting both fields * to the same non-zero value means no skew. */ db_rep->clock_skew = 1; db_rep->clock_base = 1; FLD_SET(db_rep->config, REP_C_AUTOINIT); /* * Turn on system messages by default. */ FLD_SET(dbenv->verbose, DB_VERB_REP_SYSTEM); #ifdef HAVE_REPLICATION_THREADS if ((ret = __repmgr_env_create(env, db_rep)) != 0) { __os_free(env, db_rep); return (ret); } #endif env->rep_handle = db_rep; return (0); } /* * __rep_env_destroy -- * Replication-specific destruction of the ENV structure. * * PUBLIC: void __rep_env_destroy __P((DB_ENV *)); */ void __rep_env_destroy(dbenv) DB_ENV *dbenv; { ENV *env; env = dbenv->env; if (env->rep_handle != NULL) { #ifdef HAVE_REPLICATION_THREADS __repmgr_env_destroy(env, env->rep_handle); #endif __os_free(env, env->rep_handle); env->rep_handle = NULL; } } /* * __rep_get_config -- * Return the replication subsystem configuration. * * PUBLIC: int __rep_get_config __P((DB_ENV *, u_int32_t, int *)); */ int __rep_get_config(dbenv, which, onp) DB_ENV *dbenv; u_int32_t which; int *onp; { DB_REP *db_rep; ENV *env; REP *rep; u_int32_t mapped; env = dbenv->env; #undef OK_FLAGS #define OK_FLAGS \ (DB_REP_CONF_AUTOINIT | DB_REP_CONF_BULK | DB_REP_CONF_DELAYCLIENT |\ DB_REP_CONF_INMEM | DB_REP_CONF_LEASE | DB_REP_CONF_NOWAIT | \ DB_REPMGR_CONF_2SITE_STRICT | DB_REPMGR_CONF_ELECTIONS) if (FLD_ISSET(which, ~OK_FLAGS)) return (__db_ferr(env, "DB_ENV->rep_get_config", 0)); db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_get_config", DB_INIT_REP); mapped = 0; __rep_config_map(env, &which, &mapped); if (REP_ON(env)) { rep = db_rep->region; if (FLD_ISSET(rep->config, mapped)) *onp = 1; else *onp = 0; } else { if (FLD_ISSET(db_rep->config, mapped)) *onp = 1; else *onp = 0; } return (0); } /* * __rep_set_config -- * Configure the replication subsystem. * * PUBLIC: int __rep_set_config __P((DB_ENV *, u_int32_t, int)); */ int __rep_set_config(dbenv, which, on) DB_ENV *dbenv; u_int32_t which; int on; { DB_LOG *dblp; DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; LOG *lp; REP *rep; REP_BULK bulk; u_int32_t mapped, orig; int ret, t_ret; env = dbenv->env; db_rep = env->rep_handle; ret = 0; #undef OK_FLAGS #define OK_FLAGS \ (DB_REP_CONF_AUTOINIT | DB_REP_CONF_BULK | DB_REP_CONF_DELAYCLIENT |\ DB_REP_CONF_INMEM | DB_REP_CONF_LEASE | DB_REP_CONF_NOWAIT | \ DB_REPMGR_CONF_2SITE_STRICT | DB_REPMGR_CONF_ELECTIONS) #define REPMGR_FLAGS (REP_C_2SITE_STRICT | REP_C_ELECTIONS) ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_config", DB_INIT_REP); if (FLD_ISSET(which, ~OK_FLAGS)) return (__db_ferr(env, "DB_ENV->rep_set_config", 0)); mapped = 0; __rep_config_map(env, &which, &mapped); if (APP_IS_BASEAPI(env) && FLD_ISSET(mapped, REPMGR_FLAGS)) { __db_errx(env, "%s %s", "DB_ENV->rep_set_config:", "cannot configure repmgr settings from base replication application"); return (EINVAL); } if (REP_ON(env)) { #ifdef HAVE_REPLICATION_THREADS if ((ret = __repmgr_valid_config(env, mapped)) != 0) return (ret); #endif ENV_ENTER(env, ip); rep = db_rep->region; /* * In-memory replication must be called before calling * env->open. If it is turned on and off before env->open, * it doesn't matter. Any attempt to turn it on or off after * env->open is intercepted by this error. */ if (FLD_ISSET(mapped, REP_C_INMEM)) { __db_errx(env, "%s %s", "DB_ENV->rep_set_config:", "in-memory replication must be configured before DB_ENV->open"); ENV_LEAVE(env, ip); return (EINVAL); } /* * Leases must be turned on before calling rep_start. * Leases can never be turned off once they're turned on. */ if (FLD_ISSET(mapped, REP_C_LEASE)) { if (F_ISSET(rep, REP_F_START_CALLED)) { __db_errx(env, "DB_ENV->rep_set_config: leases must be configured before DB_ENV->rep_start"); ret = EINVAL; } if (on == 0) { __db_errx(env, "DB_ENV->rep_set_config: leases cannot be turned off"); ret = EINVAL; } if (ret != 0) { ENV_LEAVE(env, ip); return (ret); } } MUTEX_LOCK(env, rep->mtx_clientdb); REP_SYSTEM_LOCK(env); orig = rep->config; if (on) FLD_SET(rep->config, mapped); else FLD_CLR(rep->config, mapped); /* * Bulk transfer requires special processing if it is getting * toggled. */ dblp = env->lg_handle; lp = dblp->reginfo.primary; if (FLD_ISSET(rep->config, REP_C_BULK) && !FLD_ISSET(orig, REP_C_BULK)) db_rep->bulk = R_ADDR(&dblp->reginfo, lp->bulk_buf); REP_SYSTEM_UNLOCK(env); /* * If turning bulk off and it was on, send out whatever is in * the buffer already. */ if (FLD_ISSET(orig, REP_C_BULK) && !FLD_ISSET(rep->config, REP_C_BULK) && lp->bulk_off != 0) { memset(&bulk, 0, sizeof(bulk)); if (db_rep->bulk == NULL) bulk.addr = R_ADDR(&dblp->reginfo, lp->bulk_buf); else bulk.addr = db_rep->bulk; bulk.offp = &lp->bulk_off; bulk.len = lp->bulk_len; bulk.type = REP_BULK_LOG; bulk.eid = DB_EID_BROADCAST; bulk.flagsp = &lp->bulk_flags; ret = __rep_send_bulk(env, &bulk, 0); } MUTEX_UNLOCK(env, rep->mtx_clientdb); ENV_LEAVE(env, ip); #ifdef HAVE_REPLICATION_THREADS /* * If turning ELECTIONS on, and it was off, check whether we * need to start an election immediately. */ if (!FLD_ISSET(orig, REP_C_ELECTIONS) && FLD_ISSET(rep->config, REP_C_ELECTIONS) && (t_ret = __repmgr_turn_on_elections(env)) != 0 && ret == 0) ret = t_ret; #endif } else { if (on) FLD_SET(db_rep->config, mapped); else FLD_CLR(db_rep->config, mapped); } /* Configuring 2SITE_STRICT, etc. makes this a repmgr application */ if (ret == 0 && FLD_ISSET(mapped, REPMGR_FLAGS)) APP_SET_REPMGR(env); return (ret); } static void __rep_config_map(env, inflagsp, outflagsp) ENV *env; u_int32_t *inflagsp, *outflagsp; { COMPQUIET(env, NULL); if (FLD_ISSET(*inflagsp, DB_REP_CONF_AUTOINIT)) { FLD_SET(*outflagsp, REP_C_AUTOINIT); FLD_CLR(*inflagsp, DB_REP_CONF_AUTOINIT); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_BULK)) { FLD_SET(*outflagsp, REP_C_BULK); FLD_CLR(*inflagsp, DB_REP_CONF_BULK); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_DELAYCLIENT)) { FLD_SET(*outflagsp, REP_C_DELAYCLIENT); FLD_CLR(*inflagsp, DB_REP_CONF_DELAYCLIENT); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_INMEM)) { FLD_SET(*outflagsp, REP_C_INMEM); FLD_CLR(*inflagsp, DB_REP_CONF_INMEM); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_LEASE)) { FLD_SET(*outflagsp, REP_C_LEASE); FLD_CLR(*inflagsp, DB_REP_CONF_LEASE); } if (FLD_ISSET(*inflagsp, DB_REP_CONF_NOWAIT)) { FLD_SET(*outflagsp, REP_C_NOWAIT); FLD_CLR(*inflagsp, DB_REP_CONF_NOWAIT); } if (FLD_ISSET(*inflagsp, DB_REPMGR_CONF_2SITE_STRICT)) { FLD_SET(*outflagsp, REP_C_2SITE_STRICT); FLD_CLR(*inflagsp, DB_REPMGR_CONF_2SITE_STRICT); } if (FLD_ISSET(*inflagsp, DB_REPMGR_CONF_ELECTIONS)) { FLD_SET(*outflagsp, REP_C_ELECTIONS); FLD_CLR(*inflagsp, DB_REPMGR_CONF_ELECTIONS); } DB_ASSERT(env, *inflagsp == 0); } /* * __rep_start_pp -- * Become a master or client, and start sending messages to participate * in the replication environment. Must be called after the environment * is open. * * PUBLIC: int __rep_start_pp __P((DB_ENV *, DBT *, u_int32_t)); */ int __rep_start_pp(dbenv, dbt, flags) DB_ENV *dbenv; DBT *dbt; u_int32_t flags; { DB_REP *db_rep; ENV *env; env = dbenv->env; db_rep = env->rep_handle; ENV_REQUIRES_CONFIG_XX( env, rep_handle, "DB_ENV->rep_start", DB_INIT_REP); if (APP_IS_REPMGR(env)) { __db_errx(env, "DB_ENV->rep_start: cannot call from Replication Manager application"); return (EINVAL); } switch (LF_ISSET(DB_REP_CLIENT | DB_REP_MASTER)) { case DB_REP_CLIENT: case DB_REP_MASTER: break; default: __db_errx(env, "DB_ENV->rep_start: must specify DB_REP_CLIENT or DB_REP_MASTER"); return (EINVAL); } /* We need a transport function because we send messages. */ if (db_rep->send == NULL) { __db_errx(env, "DB_ENV->rep_start: must be called after DB_ENV->rep_set_transport"); return (EINVAL); } return (__rep_start_int(env, dbt, flags)); } /* * __rep_start_int -- * Internal processing to become a master or client and start sending * messages to participate in the replication environment. * * We must protect rep_start_int, which may change the world, with the rest * of the DB library. Each API interface will count itself as it enters * the library. Rep_start_int checks the following: * * rep->msg_th - this is the count of threads currently in rep_process_message * rep->handle_cnt - number of threads actively using a dbp in library. * rep->txn_cnt - number of active txns. * REP_LOCKOUT_* - Replication flag that indicates that we wish to run * recovery, and want to prohibit new transactions from entering and cause * existing ones to return immediately (with a DB_LOCK_DEADLOCK error). * * There is also the renv->rep_timestamp which is updated whenever significant * events (i.e., new masters, log rollback, etc). Upon creation, a handle * is associated with the current timestamp. Each time a handle enters the * library it must check if the handle timestamp is the same as the one * stored in the replication region. This prevents the use of handles on * clients that reference non-existent files whose creation was backed out * during a synchronizing recovery. * * PUBLIC: int __rep_start_int __P((ENV *, DBT *, u_int32_t)); */ int __rep_start_int(env, dbt, flags) ENV *env; DBT *dbt; u_int32_t flags; { DB *dbp; DB_LOG *dblp; DB_LOGC *logc; DB_LSN lsn, perm_lsn; DB_REP *db_rep; DB_THREAD_INFO *ip; DB_TXNREGION *region; LOG *lp; REGENV *renv; REGINFO *infop; REP *rep; db_timeout_t tmp; u_int32_t new_gen, oldvers, pending_event, role; int interrupting, locked, ret, role_chg, start_th, t_ret; db_rep = env->rep_handle; rep = db_rep->region; infop = env->reginfo; renv = infop->primary; interrupting = locked = 0; pending_event = DB_EVENT_NO_SUCH_EVENT; role = LF_ISSET(DB_REP_CLIENT | DB_REP_MASTER); start_th = 0; /* * If we're using master leases, check that all needed * setup has been done, including setting the lease timeout. */ if (IS_USING_LEASES(env) && rep->lease_timeout == 0) { __db_errx(env, "DB_ENV->rep_start: must call DB_ENV->rep_set_timeout for leases first"); return (EINVAL); } ENV_ENTER(env, ip); /* Serialize rep_start() calls. */ MUTEX_LOCK(env, rep->mtx_repstart); start_th = 1; /* * In order to correctly check log files for old versions, we * need to flush the logs. Serialize log flush to make sure it is * always done just before the log old version check. Otherwise it * is possible that another thread in rep_start could write LSN history * and create a new log file that is not yet fully there for the log * old version check. */ if ((ret = __log_flush(env, NULL)) != 0) goto out; REP_SYSTEM_LOCK(env); role_chg = (!F_ISSET(rep, REP_F_MASTER) && role == DB_REP_MASTER) || (!F_ISSET(rep, REP_F_CLIENT) && role == DB_REP_CLIENT); /* * There is no need for lockout if all we're doing is sending a message. * In fact, lockout could be harmful: the typical use of this "duplicate * client" style of call is when the application has to poll, seeking * for a master. If the resulting NEWMASTER message were to arrive when * we had messages locked out, we would discard it, resulting in further * delay. */ if (role == DB_REP_CLIENT && !role_chg) { REP_SYSTEM_UNLOCK(env); if ((ret = __dbt_usercopy(env, dbt)) == 0) (void)__rep_send_message(env, DB_EID_BROADCAST, REP_NEWCLIENT, NULL, dbt, 0, 0); goto out; } if (FLD_ISSET(rep->lockout_flags, REP_LOCKOUT_MSG)) { /* * There is already someone in msg lockout. Return. */ RPRINT(env, (env, DB_VERB_REP_MISC, "Thread already in msg lockout")); REP_SYSTEM_UNLOCK(env); goto out; } else if ((ret = __rep_lockout_msg(env, rep, 0)) != 0) goto errunlock; /* * If we are internal init and we try to become master, reject it. * Our environment databases/logs are in an inconsistent state and * we cannot become master. */ if (IN_INTERNAL_INIT(rep) && role == DB_REP_MASTER) { __db_errx(env, "DB_ENV->rep_start: Cannot become master during internal init"); ret = DB_REP_UNAVAIL; goto errunlock; } /* * Wait for any active txns or mpool ops to complete, and * prevent any new ones from occurring, only if we're * changing roles. */ if (role_chg) { if ((ret = __rep_lockout_api(env, rep)) != 0) goto errunlock; locked = 1; } dblp = env->lg_handle; lp = dblp->reginfo.primary; if (role == DB_REP_MASTER) { if (role_chg) { /* * If we were previously a client, it's possible we * could have an interruptible STARTSYNC in progress. * Interrupt it now, so that it doesn't slow down our * transition to master, and because its effects aren't * doing us any good anyway. */ (void)__memp_set_config( env->dbenv, DB_MEMP_SYNC_INTERRUPT, 1); interrupting = 1; /* * If we're upgrading from having been a client, * preclose, so that we close our temporary database * and any files we opened while doing a rep_apply. * If we don't we can infinitely leak file ids if * the master crashed with files open (the likely * case). If we don't close them we can run into * problems if we try to remove that file or long * running applications end up with an unbounded * number of used fileids, each getting written * on checkpoint. Just close them. * Then invalidate all files open in the logging * region. These are files open by other processes * attached to the environment. They must be * closed by the other processes when they notice * the change in role. */ if ((ret = __rep_preclose(env)) != 0) goto errunlock; new_gen = rep->gen + 1; /* * There could have been any number of failed * elections, so jump the gen if we need to now. */ if (rep->egen > rep->gen) new_gen = rep->egen; SET_GEN(new_gen); if (IS_USING_LEASES(env) && !F_ISSET(rep, REP_F_MASTERELECT)) { __db_errx(env, "rep_start: Cannot become master without being elected when using leases."); ret = EINVAL; goto errunlock; } if (F_ISSET(rep, REP_F_MASTERELECT)) { __rep_elect_done(env, rep); F_CLR(rep, REP_F_MASTERELECT); } else if (FLD_ISSET(rep->config, REP_C_INMEM)) /* * Help detect if application has ignored our * recommendation against reappointing same * master after a crash/reboot when running * in-memory replication. Doing this allows a * slight chance of two masters at the same * generation, resulting in client crashes. */ RPRINT(env, (env, DB_VERB_REP_MISC, "Appointed new master while running in-memory replication.")); if (rep->egen <= rep->gen) rep->egen = rep->gen + 1; RPRINT(env, (env, DB_VERB_REP_MISC, "New master gen %lu, egen %lu", (u_long)rep->gen, (u_long)rep->egen)); /* * If not running in-memory replication, write * gen file. */ if (!FLD_ISSET(rep->config, REP_C_INMEM) && (ret = __rep_write_gen(env, rep, rep->gen)) != 0) goto errunlock; } /* * Set lease duration assuming clients have faster clock. * Master needs to compensate so that clients do not * expire their grant while the master thinks it is valid. */ if (IS_USING_LEASES(env) && (role_chg || !IS_REP_STARTED(env))) { /* * If we have already granted our lease, we * cannot become master. */ if ((ret = __rep_islease_granted(env))) { __db_errx(env, "rep_start: Cannot become master with outstanding lease granted."); ret = EINVAL; goto errunlock; } /* * Set max_perm_lsn to last PERM record on master. */ if ((ret = __log_cursor(env, &logc)) != 0) goto errunlock; ret = __rep_log_backup(env, rep, logc, &perm_lsn); (void)__logc_close(logc); /* * If we found a perm LSN use it. Otherwise, if * no perm LSN exists, initialize. */ if (ret == 0) lp->max_perm_lsn = perm_lsn; else if (ret == DB_NOTFOUND) INIT_LSN(lp->max_perm_lsn); else goto errunlock; /* * Simply compute the larger ratio for the lease. */ tmp = (db_timeout_t)((double)rep->lease_timeout / ((double)rep->clock_skew / (double)rep->clock_base)); DB_TIMEOUT_TO_TIMESPEC(tmp, &rep->lease_duration); if ((ret = __rep_lease_table_alloc(env, rep->nsites)) != 0) goto errunlock; } rep->master_id = rep->eid; STAT_INC(env, rep, master_change, rep->stat.st_master_changes, rep->eid); #ifdef DIAGNOSTIC if (!F_ISSET(rep, REP_F_GROUP_ESTD)) RPRINT(env, (env, DB_VERB_REP_MISC, "Establishing group as master.")); #endif /* * When becoming a master, clear the following flags: * CLIENT: Site is no longer a client. * ABBREVIATED: Indicates abbreviated internal init, which * cannot occur on a master. * MASTERELECT: Indicates that this master is elected * rather than appointed. If we're changing roles we * used this flag above for error checks and election * cleanup. * SKIPPED_APPLY: Indicates that client apply skipped * some log records during an election, no longer * applicable on master. * DELAY: Indicates user config to delay initial client * sync with new master, doesn't apply to master. * LEASE_EXPIRED: Applies to client leases which are * now defunct on master. * NEWFILE: Used to delay client apply during newfile * operation, not applicable to master. */ F_CLR(rep, REP_F_CLIENT | REP_F_ABBREVIATED | REP_F_MASTERELECT | REP_F_SKIPPED_APPLY | REP_F_DELAY | REP_F_LEASE_EXPIRED | REP_F_NEWFILE); /* * When becoming a master, set the following flags: * MASTER: Indicate that this site is master. * GROUP_ESTD: Having a master means a that replication * group exists. * NIMDBS_LOADED: Inmem dbs are always present on a master. */ F_SET(rep, REP_F_MASTER | REP_F_GROUP_ESTD | REP_F_NIMDBS_LOADED); /* Master cannot be in internal init. */ rep->sync_state = SYNC_OFF; /* * We're master. Set the versions to the current ones. */ oldvers = lp->persist.version; /* * If we're moving forward to the current version, we need * to force the log file to advance and reset the * recovery table since it contains pointers to old * recovery functions. */ VPRINT(env, (env, DB_VERB_REP_MISC, "rep_start: Old log version was %lu", (u_long)oldvers)); if (lp->persist.version != DB_LOGVERSION) { if ((ret = __env_init_rec(env, DB_LOGVERSION)) != 0) goto errunlock; } rep->version = DB_REPVERSION; /* * When becoming a master, clear the following lockouts: * ARCHIVE: Used to keep logs while client may be * inconsistent, not needed on master. * MSG: We set this above to block message processing while * becoming a master, can turn messages back on here. */ FLD_CLR(rep->lockout_flags, REP_LOCKOUT_ARCHIVE | REP_LOCKOUT_MSG); REP_SYSTEM_UNLOCK(env); LOG_SYSTEM_LOCK(env); lsn = lp->lsn; LOG_SYSTEM_UNLOCK(env); /* * Send the NEWMASTER message first so that clients know * subsequent messages are coming from the right master. * We need to perform all actions below no matter what * regarding errors. */ (void)__rep_send_message(env, DB_EID_BROADCAST, REP_NEWMASTER, &lsn, NULL, 0, 0); ret = 0; if (role_chg) { pending_event = DB_EVENT_REP_MASTER; /* * If prepared transactions have not been restored * look to see if there are any. If there are, * then mark the open files, otherwise close them. */ region = env->tx_handle->reginfo.primary; if (region->stat.st_nrestores == 0 && (t_ret = __rep_restore_prepared(env)) != 0 && ret == 0) ret = t_ret; if (region->stat.st_nrestores != 0) { if ((t_ret = __dbreg_mark_restored(env)) != 0 && ret == 0) ret = t_ret; } else { ret = __dbreg_invalidate_files(env, 0); if ((t_ret = __rep_closefiles(env)) != 0 && ret == 0) ret = t_ret; } if ((t_ret = __txn_recycle_id(env)) != 0 && ret == 0) ret = t_ret; /* * Write LSN history database, ahead of unlocking the * API so that clients can always know the heritage of * any transaction they receive via replication. */ if ((t_ret = __rep_save_lsn_hist(env, ip, &lsn)) != 0 && ret == 0) ret = t_ret; REP_SYSTEM_LOCK(env); rep->gen_base_lsn = lsn; rep->master_envid = renv->envid; CLR_LOCKOUT_BDB(rep); locked = 0; REP_SYSTEM_UNLOCK(env); (void)__memp_set_config( env->dbenv, DB_MEMP_SYNC_INTERRUPT, 0); interrupting = 0; } } else { /* * Start a non-client as a client. */ rep->master_id = DB_EID_INVALID; /* * A non-client should not have been participating in an * election, so most election flags should be off. The TALLY * flag is an exception because it is set any time we receive * a VOTE1 and there is no reason to clear and lose it for an * election that may begin shortly. */ DB_ASSERT(env, !FLD_ISSET(rep->elect_flags, ~REP_E_TALLY)); /* * A non-client should not have the following client flags * set and should not be in internal init. */ DB_ASSERT(env, !F_ISSET(rep, REP_F_ABBREVIATED | REP_F_DELAY | REP_F_NEWFILE)); DB_ASSERT(env, rep->sync_state == SYNC_OFF); if ((ret = __log_get_oldversion(env, &oldvers)) != 0) goto errunlock; RPRINT(env, (env, DB_VERB_REP_MISC, "rep_start: Found old version log %d", oldvers)); if (oldvers >= DB_LOGVERSION_MIN) { __log_set_version(env, oldvers); if ((ret = __env_init_rec(env, oldvers)) != 0) goto errunlock; oldvers = __rep_conv_vers(env, oldvers); DB_ASSERT(env, oldvers != DB_REPVERSION_INVALID); rep->version = oldvers; } /* * When becoming a client, clear the following flags: * MASTER: Site is no longer a master. * MASTERELECT: Indicates that a master is elected * rather than appointed, not applicable on client. */ F_CLR(rep, REP_F_MASTER | REP_F_MASTERELECT); F_SET(rep, REP_F_CLIENT); /* * On a client, compute the lease duration on the * assumption that the client has a fast clock. * Expire any existing leases we might have held as * a master. */ if (IS_USING_LEASES(env) && !IS_REP_STARTED(env)) { if ((ret = __rep_lease_expire(env)) != 0) goto errunlock; /* * Since the master is also compensating on its * side as well, we're being doubly conservative * to compensate on the client side. Theoretically, * this compensation is not necessary, as it is * effectively doubling the skew compensation. * But we are making guarantees based on time and * skews across machines. So we are being extra * cautious. */ tmp = (db_timeout_t)((double)rep->lease_timeout * ((double)rep->clock_skew / (double)rep->clock_base)); DB_TIMEOUT_TO_TIMESPEC(tmp, &rep->lease_duration); if (rep->lease_off != INVALID_ROFF) { MUTEX_LOCK(env, renv->mtx_regenv); __env_alloc_free(infop, R_ADDR(infop, rep->lease_off)); MUTEX_UNLOCK(env, renv->mtx_regenv); rep->lease_off = INVALID_ROFF; } } REP_SYSTEM_UNLOCK(env); /* * Abort any prepared transactions that were restored * by recovery. We won't be able to create any txns of * our own until they're resolved, but we can't resolve * them ourselves; the master has to. If any get * resolved as commits, we'll redo them when commit * records come in. Aborts will simply be ignored. */ if ((ret = __rep_abort_prepared(env)) != 0) goto errlock; /* * Since we're changing roles we need to init the db. */ if ((ret = __db_create_internal(&dbp, env, 0)) != 0) goto errlock; /* * Ignore errors, because if the file doesn't exist, * this is perfectly OK. */ MUTEX_LOCK(env, rep->mtx_clientdb); (void)__db_remove(dbp, ip, NULL, REPDBNAME, NULL, DB_FORCE); MUTEX_UNLOCK(env, rep->mtx_clientdb); /* * Set pending_event after calls that can fail. */ pending_event = DB_EVENT_REP_CLIENT; REP_SYSTEM_LOCK(env); FLD_CLR(rep->lockout_flags, REP_LOCKOUT_MSG); if (locked) { CLR_LOCKOUT_BDB(rep); locked = 0; } if (F_ISSET(env, ENV_PRIVATE)) /* * If we think we're a new client, and we have a * private env, set our gen number down to 0. * Otherwise, we can restart and think * we're ready to accept a new record (because our * gen is okay), but really this client needs to * sync with the master. */ SET_GEN(0); REP_SYSTEM_UNLOCK(env); /* * Announce ourselves and send out our data. */ if ((ret = __dbt_usercopy(env, dbt)) != 0) goto out; (void)__rep_send_message(env, DB_EID_BROADCAST, REP_NEWCLIENT, NULL, dbt, 0, 0); } if (0) { /* * We have separate labels for errors. If we're returning an * error before we've set REP_LOCKOUT_MSG, we use 'err'. If * we are erroring while holding the region mutex, then we use * 'errunlock' label. If we error without holding the rep * mutex we must use 'errlock'. */ errlock: REP_SYSTEM_LOCK(env); errunlock: FLD_CLR(rep->lockout_flags, REP_LOCKOUT_MSG); if (locked) CLR_LOCKOUT_BDB(rep); if (interrupting) (void)__memp_set_config( env->dbenv, DB_MEMP_SYNC_INTERRUPT, 0); REP_SYSTEM_UNLOCK(env); } out: if (ret == 0) { REP_SYSTEM_LOCK(env); F_SET(rep, REP_F_START_CALLED); REP_SYSTEM_UNLOCK(env); } if (start_th) MUTEX_UNLOCK(env, rep->mtx_repstart); if (pending_event != DB_EVENT_NO_SUCH_EVENT) __rep_fire_event(env, pending_event, NULL); __dbt_userfree(env, dbt, NULL, NULL); ENV_LEAVE(env, ip); return (ret); } /* * Write the current generation's base LSN into the history database. */ static int __rep_save_lsn_hist(env, ip, lsnp) ENV *env; DB_THREAD_INFO *ip; DB_LSN *lsnp; { DB_REP *db_rep; REP *rep; REGENV *renv; DB_TXN *txn; DB *dbp; DBT key_dbt, data_dbt; __rep_lsn_hist_key_args key; __rep_lsn_hist_data_args data; u_int8_t key_buf[__REP_LSN_HIST_KEY_SIZE]; u_int8_t data_buf[__REP_LSN_HIST_DATA_SIZE]; db_timespec now; int ret, t_ret; db_rep = env->rep_handle; rep = db_rep->region; renv = env->reginfo->primary; txn = NULL; ret = 0; if ((ret = __txn_begin(env, ip, NULL, &txn, DB_IGNORE_LEASE)) != 0) return (ret); /* * Use the cached handle to the history database if it is already open. * Since we're becoming master, we don't expect to need it after this, * so clear the cached handle and close the database once we've written * our update. */ if ((dbp = db_rep->lsn_db) == NULL) { if ((ret = __rep_open_lsn_history(env, ip, txn, DB_CREATE, &dbp)) != 0) goto err; } else db_rep->lsn_db = NULL; key.version = REP_LSN_HISTORY_FMT_VERSION; key.gen = rep->gen; __rep_lsn_hist_key_marshal(env, &key, key_buf); data.envid = renv->envid; data.lsn = *lsnp; __os_gettime(env, &now, 0); data.hist_sec = (u_int32_t)now.tv_sec; data.hist_nsec = (u_int32_t)now.tv_nsec; __rep_lsn_hist_data_marshal(env, &data, data_buf); DB_INIT_DBT(key_dbt, key_buf, sizeof(key_buf)); DB_INIT_DBT(data_dbt, data_buf, sizeof(data_buf)); ret = __db_put(dbp, ip, txn, &key_dbt, &data_dbt, 0); err: if (dbp != NULL && (t_ret = __db_close(dbp, txn, DB_NOSYNC)) != 0 && ret == 0) ret = t_ret; DB_ASSERT(env, txn != NULL); if ((t_ret = __db_txn_auto_resolve(env, txn, 0, ret)) && ret == 0) ret = t_ret; return (ret); } /* * Open existing LSN history database, wherever it may be (on disk or in * memory). If it doesn't exist, create it only if DB_CREATE is specified by * our caller. * * If we could be sure that all sites in the replication group had matching * REP_C_INMEM settings (that never changed over time), we could simply look for * the database in the place where we knew it should be. The code here tries to * be more flexible/resilient to mis-matching INMEM settings, even though we * recommend against that. */ static int __rep_open_lsn_history(env, ip, txn, flags, dbpp) ENV *env; DB_THREAD_INFO *ip; DB_TXN *txn; u_int32_t flags; DB **dbpp; { DB_REP *db_rep; REP *rep; DB *dbp; char *fname; u_int32_t myflags; int ret, t_ret; db_rep = env->rep_handle; rep = db_rep->region; if ((ret = __db_create_internal(&dbp, env, 0)) != 0) return (ret); myflags = (F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0); /* * First, try opening it as a sub-database within a disk-resident * database file. (If success, skip to the end.) */ if ((ret = __db_open(dbp, ip, txn, REPSYSDBNAME, REPLSNHIST, DB_BTREE, myflags, 0, PGNO_BASE_MD)) == 0) goto found; if (ret != ENOENT) goto err; /* * Here, the file was not found. Next, try opening it as an in-memory * database (after the necessary clean-up). */ ret = __db_close(dbp, txn, DB_NOSYNC); dbp = NULL; if (ret != 0 || (ret = __db_create_internal(&dbp, env, 0)) != 0) goto err; if ((ret = __db_open(dbp, ip, txn, NULL, REPLSNHIST, DB_BTREE, myflags, 0, PGNO_BASE_MD)) == 0) goto found; if (ret != ENOENT) goto err; /* * Here, the database was not found either on disk or in memory. Create * it, according to our local INMEM setting. */ ret = __db_close(dbp, txn, DB_NOSYNC); dbp = NULL; if (ret != 0) goto err; if (LF_ISSET(DB_CREATE)) { if ((ret = __db_create_internal(&dbp, env, 0)) != 0) goto err; if ((ret = __db_set_pagesize(dbp, REPSYSDBPGSZ)) != 0) goto err; FLD_SET(myflags, DB_CREATE); fname = FLD_ISSET(rep->config, REP_C_INMEM) ? NULL : REPSYSDBNAME; if ((ret = __db_open(dbp, ip, txn, fname, REPLSNHIST, DB_BTREE, myflags, 0, PGNO_BASE_MD)) == 0) goto found; } else ret = ENOENT; err: if (dbp != NULL && (t_ret = __db_close(dbp, txn, DB_NOSYNC)) != 0 && (ret == 0 || ret == ENOENT)) ret = t_ret; return (ret); found: *dbpp = dbp; return (0); } /* * __rep_client_dbinit -- * * Initialize the LSN database on the client side. This is called from the * client initialization code. The startup flag value indicates if * this is the first thread/process starting up and therefore should create * the LSN database. This routine must be called once by each process acting * as a client. * * Assumes caller holds appropriate mutex. * * PUBLIC: int __rep_client_dbinit __P((ENV *, int, repdb_t)); */ int __rep_client_dbinit(env, startup, which) ENV *env; int startup; repdb_t which; { DB *dbp, **rdbpp; DB_REP *db_rep; DB_THREAD_INFO *ip; REP *rep; int ret, t_ret; u_int32_t flags; const char *fname, *name, *subdb; db_rep = env->rep_handle; rep = db_rep->region; dbp = NULL; if (which == REP_DB) { name = REPDBNAME; rdbpp = &db_rep->rep_db; } else { name = REPPAGENAME; rdbpp = &db_rep->file_dbp; } /* Check if this has already been called on this environment. */ if (*rdbpp != NULL) return (0); ENV_GET_THREAD_INFO(env, ip); /* Set up arguments for __db_remove and __db_open calls. */ fname = name; subdb = NULL; if (FLD_ISSET(rep->config, REP_C_INMEM)) { fname = NULL; subdb = name; } if (startup) { if ((ret = __db_create_internal(&dbp, env, 0)) != 0) goto err; /* * Prevent in-memory database remove from writing to * non-existent logs. */ if (FLD_ISSET(rep->config, REP_C_INMEM)) (void)__db_set_flags(dbp, DB_TXN_NOT_DURABLE); /* * Ignore errors, because if the file doesn't exist, this * is perfectly OK. */ (void)__db_remove(dbp, ip, NULL, fname, subdb, DB_FORCE); } if ((ret = __db_create_internal(&dbp, env, 0)) != 0) goto err; if (which == REP_DB && (ret = __bam_set_bt_compare(dbp, __rep_bt_cmp)) != 0) goto err; /* Don't write log records on the client. */ if ((ret = __db_set_flags(dbp, DB_TXN_NOT_DURABLE)) != 0) goto err; flags = DB_NO_AUTO_COMMIT | DB_CREATE | (F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0); if ((ret = __db_open(dbp, ip, NULL, fname, subdb, (which == REP_DB ? DB_BTREE : DB_RECNO), flags, 0, PGNO_BASE_MD)) != 0) goto err; *rdbpp = dbp; if (0) { err: if (dbp != NULL && (t_ret = __db_close(dbp, NULL, DB_NOSYNC)) != 0 && ret == 0) ret = t_ret; *rdbpp = NULL; } return (ret); } /* * __rep_bt_cmp -- * * Comparison function for the LSN table. We use the entire control * structure as a key (for simplicity, so we don't have to merge the * other fields in the control with the data field), but really only * care about the LSNs. */ static int __rep_bt_cmp(dbp, dbt1, dbt2) DB *dbp; const DBT *dbt1, *dbt2; { DB_LSN lsn1, lsn2; __rep_control_args *rp1, *rp2; COMPQUIET(dbp, NULL); rp1 = dbt1->data; rp2 = dbt2->data; (void)__ua_memcpy(&lsn1, &rp1->lsn, sizeof(DB_LSN)); (void)__ua_memcpy(&lsn2, &rp2->lsn, sizeof(DB_LSN)); if (lsn1.file > lsn2.file) return (1); if (lsn1.file < lsn2.file) return (-1); if (lsn1.offset > lsn2.offset) return (1); if (lsn1.offset < lsn2.offset) return (-1); return (0); } /* * __rep_abort_prepared -- * Abort any prepared transactions that recovery restored. * * This is used by clients that have just run recovery, since * they cannot/should not call txn_recover and handle prepared transactions * themselves. */ static int __rep_abort_prepared(env) ENV *env; { #define PREPLISTSIZE 50 DB_LOG *dblp; DB_PREPLIST prep[PREPLISTSIZE], *p; DB_TXNMGR *mgr; DB_TXNREGION *region; LOG *lp; int ret; u_int32_t count, i; u_int32_t op; mgr = env->tx_handle; region = mgr->reginfo.primary; dblp = env->lg_handle; lp = dblp->reginfo.primary; if (region->stat.st_nrestores == 0) return (0); op = DB_FIRST; do { if ((ret = __txn_recover(env, prep, PREPLISTSIZE, &count, op)) != 0) return (ret); for (i = 0; i < count; i++) { p = &prep[i]; if ((ret = __txn_abort(p->txn)) != 0) return (ret); env->rep_handle->region->op_cnt--; env->rep_handle->region->max_prep_lsn = lp->lsn; region->stat.st_nrestores--; } op = DB_NEXT; } while (count == PREPLISTSIZE); return (0); } /* * __rep_restore_prepared -- * Restore to a prepared state any prepared but not yet committed * transactions. * * This performs, in effect, a "mini-recovery"; it is called from * __rep_start by newly upgraded masters. There may be transactions that an * old master prepared but did not resolve, which we need to restore to an * active state. */ static int __rep_restore_prepared(env) ENV *env; { DBT rec; DB_LOGC *logc; DB_LSN ckp_lsn, lsn; DB_REP *db_rep; DB_TXNHEAD *txninfo; REP *rep; __txn_ckp_args *ckp_args; __txn_regop_args *regop_args; __txn_prepare_args *prep_args; int ret, t_ret; u_int32_t hi_txn, low_txn, rectype, status, txnid, txnop; db_rep = env->rep_handle; rep = db_rep->region; if (IS_ZERO_LSN(rep->max_prep_lsn)) { VPRINT(env, (env, DB_VERB_REP_MISC, "restore_prep: No prepares. Skip.")); return (0); } txninfo = NULL; ckp_args = NULL; prep_args = NULL; regop_args = NULL; ZERO_LSN(ckp_lsn); ZERO_LSN(lsn); if ((ret = __log_cursor(env, &logc)) != 0) return (ret); /* * Get our first LSN to see if the prepared LSN is still * available. If so, it might be unresolved. If not, * then it is guaranteed to be resolved. */ memset(&rec, 0, sizeof(DBT)); if ((ret = __logc_get(logc, &lsn, &rec, DB_FIRST)) != 0) { __db_errx(env, "First record not found"); goto err; } /* * If the max_prep_lsn is no longer available, we're sure * that txn has been resolved. We're done. */ if (rep->max_prep_lsn.file < lsn.file) { VPRINT(env, (env, DB_VERB_REP_MISC, "restore_prep: Prepare resolved. Skip")); ZERO_LSN(rep->max_prep_lsn); goto done; } /* * We need to consider the set of records between the most recent * checkpoint LSN and the end of the log; any txn in that * range, and only txns in that range, could still have been * active, and thus prepared but not yet committed (PBNYC), * when the old master died. * * Find the most recent checkpoint LSN, and get the record there. * If there is no checkpoint in the log, start off by getting * the very first record in the log instead. */ if ((ret = __txn_getckp(env, &lsn)) == 0) { if ((ret = __logc_get(logc, &lsn, &rec, DB_SET)) != 0) { __db_errx(env, "Checkpoint record at LSN [%lu][%lu] not found", (u_long)lsn.file, (u_long)lsn.offset); goto err; } if ((ret = __txn_ckp_read( env, rec.data, &ckp_args)) == 0) { ckp_lsn = ckp_args->ckp_lsn; __os_free(env, ckp_args); } if (ret != 0) { __db_errx(env, "Invalid checkpoint record at [%lu][%lu]", (u_long)lsn.file, (u_long)lsn.offset); goto err; } if ((ret = __logc_get(logc, &ckp_lsn, &rec, DB_SET)) != 0) { __db_errx(env, "Checkpoint LSN record [%lu][%lu] not found", (u_long)ckp_lsn.file, (u_long)ckp_lsn.offset); goto err; } } else if ((ret = __logc_get(logc, &lsn, &rec, DB_FIRST)) != 0) { if (ret == DB_NOTFOUND) { /* An empty log means no PBNYC txns. */ ret = 0; goto done; } __db_errx(env, "Attempt to get first log record failed"); goto err; } /* * We use the same txnlist infrastructure that recovery does; * it demands an estimate of the high and low txnids for * initialization. * * First, the low txnid. */ do { /* txnid is after rectype, which is a u_int32. */ LOGCOPY_32(env, &low_txn, (u_int8_t *)rec.data + sizeof(u_int32_t)); if (low_txn != 0) break; } while ((ret = __logc_get(logc, &lsn, &rec, DB_NEXT)) == 0); /* If there are no txns, there are no PBNYC txns. */ if (ret == DB_NOTFOUND) { ret = 0; goto done; } else if (ret != 0) goto err; /* Now, the high txnid. */ if ((ret = __logc_get(logc, &lsn, &rec, DB_LAST)) != 0) { /* * Note that DB_NOTFOUND is unacceptable here because we * had to have looked at some log record to get this far. */ __db_errx(env, "Final log record not found"); goto err; } do { /* txnid is after rectype, which is a u_int32. */ LOGCOPY_32(env, &hi_txn, (u_int8_t *)rec.data + sizeof(u_int32_t)); if (hi_txn != 0) break; } while ((ret = __logc_get(logc, &lsn, &rec, DB_PREV)) == 0); if (ret == DB_NOTFOUND) { ret = 0; goto done; } else if (ret != 0) goto err; /* We have a high and low txnid. Initialise the txn list. */ if ((ret = __db_txnlist_init(env, NULL, low_txn, hi_txn, NULL, &txninfo)) != 0) goto err; /* * Now, walk backward from the end of the log to ckp_lsn. Any * prepares that we hit without first hitting a commit or * abort belong to PBNYC txns, and we need to apply them and * restore them to a prepared state. * * Note that we wind up applying transactions out of order. * Since all PBNYC txns still held locks on the old master and * were isolated, this should be safe. */ F_SET(env->lg_handle, DBLOG_RECOVER); for (ret = __logc_get(logc, &lsn, &rec, DB_LAST); ret == 0 && LOG_COMPARE(&lsn, &ckp_lsn) > 0; ret = __logc_get(logc, &lsn, &rec, DB_PREV)) { LOGCOPY_32(env, &rectype, rec.data); switch (rectype) { case DB___txn_regop: /* * It's a commit or abort--but we don't care * which! Just add it to the list of txns * that are resolved. */ if ((ret = __txn_regop_read( env, rec.data, ®op_args)) != 0) goto err; txnid = regop_args->txnp->txnid; txnop = regop_args->opcode; __os_free(env, regop_args); ret = __db_txnlist_find(env, txninfo, txnid, &status); if (ret == DB_NOTFOUND) ret = __db_txnlist_add(env, txninfo, txnid, txnop, &lsn); else if (ret != 0) goto err; break; case DB___txn_prepare: /* * It's a prepare. If its not aborted and * we haven't put the txn on our list yet, it * hasn't been resolved, so apply and restore it. */ if ((ret = __txn_prepare_read( env, rec.data, &prep_args)) != 0) goto err; ret = __db_txnlist_find(env, txninfo, prep_args->txnp->txnid, &status); if (ret == DB_NOTFOUND) { if (prep_args->opcode == TXN_ABORT) ret = __db_txnlist_add(env, txninfo, prep_args->txnp->txnid, prep_args->opcode, &lsn); else if ((ret = __rep_process_txn(env, &rec)) == 0) { /* * We are guaranteed to be single * threaded here. We need to * account for this newly * instantiated txn in the op_cnt * so that it is counted when it is * resolved. */ rep->op_cnt++; ret = __txn_restore_txn(env, &lsn, prep_args); } } else if (ret != 0) goto err; __os_free(env, prep_args); break; default: continue; } } /* It's not an error to have hit the beginning of the log. */ if (ret == DB_NOTFOUND) ret = 0; done: err: t_ret = __logc_close(logc); F_CLR(env->lg_handle, DBLOG_RECOVER); if (txninfo != NULL) __db_txnlist_end(env, txninfo); return (ret == 0 ? t_ret : ret); } /* * __rep_get_limit -- * Get the limit on the amount of data that will be sent during a single * invocation of __rep_process_message. * * PUBLIC: int __rep_get_limit __P((DB_ENV *, u_int32_t *, u_int32_t *)); */ int __rep_get_limit(dbenv, gbytesp, bytesp) DB_ENV *dbenv; u_int32_t *gbytesp, *bytesp; { DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_get_limit", DB_INIT_REP); if (REP_ON(env)) { rep = db_rep->region; ENV_ENTER(env, ip); REP_SYSTEM_LOCK(env); if (gbytesp != NULL) *gbytesp = rep->gbytes; if (bytesp != NULL) *bytesp = rep->bytes; REP_SYSTEM_UNLOCK(env); ENV_LEAVE(env, ip); } else { if (gbytesp != NULL) *gbytesp = db_rep->gbytes; if (bytesp != NULL) *bytesp = db_rep->bytes; } return (0); } /* * __rep_set_limit -- * Set a limit on the amount of data that will be sent during a single * invocation of __rep_process_message. * * PUBLIC: int __rep_set_limit __P((DB_ENV *, u_int32_t, u_int32_t)); */ int __rep_set_limit(dbenv, gbytes, bytes) DB_ENV *dbenv; u_int32_t gbytes, bytes; { DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_limit", DB_INIT_REP); if (bytes > GIGABYTE) { gbytes += bytes / GIGABYTE; bytes = bytes % GIGABYTE; } if (REP_ON(env)) { rep = db_rep->region; ENV_ENTER(env, ip); REP_SYSTEM_LOCK(env); rep->gbytes = gbytes; rep->bytes = bytes; REP_SYSTEM_UNLOCK(env); ENV_LEAVE(env, ip); } else { db_rep->gbytes = gbytes; db_rep->bytes = bytes; } return (0); } /* * PUBLIC: int __rep_set_nsites __P((DB_ENV *, u_int32_t)); */ int __rep_set_nsites(dbenv, n) DB_ENV *dbenv; u_int32_t n; { DB_REP *db_rep; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_nsites", DB_INIT_REP); if (IS_USING_LEASES(env) && IS_REP_STARTED(env)) { __db_errx(env, "DB_ENV->rep_set_nsites: must be called before DB_ENV->rep_start"); return (EINVAL); } if (REP_ON(env)) { rep = db_rep->region; rep->config_nsites = n; } else db_rep->config_nsites = n; return (0); } /* * PUBLIC: int __rep_get_nsites __P((DB_ENV *, u_int32_t *)); */ int __rep_get_nsites(dbenv, n) DB_ENV *dbenv; u_int32_t *n; { DB_REP *db_rep; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_get_nsites", DB_INIT_REP); if (REP_ON(env)) { rep = db_rep->region; *n = rep->config_nsites; } else *n = db_rep->config_nsites; return (0); } /* * PUBLIC: int __rep_set_priority __P((DB_ENV *, u_int32_t)); */ int __rep_set_priority(dbenv, priority) DB_ENV *dbenv; u_int32_t priority; { DB_REP *db_rep; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_priority", DB_INIT_REP); if (REP_ON(env)) { rep = db_rep->region; rep->priority = priority; } else db_rep->my_priority = priority; return (0); } /* * PUBLIC: int __rep_get_priority __P((DB_ENV *, u_int32_t *)); */ int __rep_get_priority(dbenv, priority) DB_ENV *dbenv; u_int32_t *priority; { DB_REP *db_rep; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_get_priority", DB_INIT_REP); if (REP_ON(env)) { rep = db_rep->region; *priority = rep->priority; } else *priority = db_rep->my_priority; return (0); } /* * PUBLIC: int __rep_set_timeout __P((DB_ENV *, int, db_timeout_t)); */ int __rep_set_timeout(dbenv, which, timeout) DB_ENV *dbenv; int which; db_timeout_t timeout; { DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; REP *rep; int repmgr_timeout, ret; env = dbenv->env; db_rep = env->rep_handle; rep = db_rep->region; ret = 0; repmgr_timeout = 0; if (timeout == 0 && (which == DB_REP_CONNECTION_RETRY || which == DB_REP_ELECTION_TIMEOUT || which == DB_REP_LEASE_TIMEOUT || which == DB_REP_ELECTION_RETRY)) { __db_errx(env, "timeout value must be > 0"); return (EINVAL); } if (which == DB_REP_ACK_TIMEOUT || which == DB_REP_CONNECTION_RETRY || which == DB_REP_ELECTION_RETRY || which == DB_REP_HEARTBEAT_MONITOR || which == DB_REP_HEARTBEAT_SEND) repmgr_timeout = 1; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_timeout", DB_INIT_REP); if (APP_IS_BASEAPI(env) && repmgr_timeout) { __db_errx(env, "%s %s", "DB_ENV->rep_set_timeout:", "cannot set Replication Manager timeout from base replication application"); return (EINVAL); } if (which == DB_REP_LEASE_TIMEOUT && IS_REP_STARTED(env)) { ret = EINVAL; __db_errx(env, "%s %s", "DB_ENV->rep_set_timeout:", "lease timeout must be set before DB_ENV->rep_start."); return (EINVAL); } switch (which) { case DB_REP_CHECKPOINT_DELAY: if (REP_ON(env)) rep->chkpt_delay = timeout; else db_rep->chkpt_delay = timeout; break; case DB_REP_ELECTION_TIMEOUT: if (REP_ON(env)) rep->elect_timeout = timeout; else db_rep->elect_timeout = timeout; break; case DB_REP_FULL_ELECTION_TIMEOUT: if (REP_ON(env)) rep->full_elect_timeout = timeout; else db_rep->full_elect_timeout = timeout; break; case DB_REP_LEASE_TIMEOUT: if (REP_ON(env)) rep->lease_timeout = timeout; else db_rep->lease_timeout = timeout; break; #ifdef HAVE_REPLICATION_THREADS case DB_REP_ACK_TIMEOUT: if (REP_ON(env)) rep->ack_timeout = timeout; else db_rep->ack_timeout = timeout; break; case DB_REP_CONNECTION_RETRY: if (REP_ON(env)) rep->connection_retry_wait = timeout; else db_rep->connection_retry_wait = timeout; break; case DB_REP_ELECTION_RETRY: if (REP_ON(env)) rep->election_retry_wait = timeout; else db_rep->election_retry_wait = timeout; break; case DB_REP_HEARTBEAT_MONITOR: if (REP_ON(env)) rep->heartbeat_monitor_timeout = timeout; else db_rep->heartbeat_monitor_timeout = timeout; break; case DB_REP_HEARTBEAT_SEND: if (REP_ON(env)) rep->heartbeat_frequency = timeout; else db_rep->heartbeat_frequency = timeout; break; #endif default: __db_errx(env, "Unknown timeout type argument to DB_ENV->rep_set_timeout"); ret = EINVAL; } /* Setting a repmgr timeout makes this a repmgr application */ if (ret == 0 && repmgr_timeout) APP_SET_REPMGR(env); return (ret); } /* * PUBLIC: int __rep_get_timeout __P((DB_ENV *, int, db_timeout_t *)); */ int __rep_get_timeout(dbenv, which, timeout) DB_ENV *dbenv; int which; db_timeout_t *timeout; { DB_REP *db_rep; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; rep = db_rep->region; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_get_timeout", DB_INIT_REP); switch (which) { case DB_REP_CHECKPOINT_DELAY: *timeout = REP_ON(env) ? rep->chkpt_delay : db_rep->chkpt_delay; break; case DB_REP_ELECTION_TIMEOUT: *timeout = REP_ON(env) ? rep->elect_timeout : db_rep->elect_timeout; break; case DB_REP_FULL_ELECTION_TIMEOUT: *timeout = REP_ON(env) ? rep->full_elect_timeout : db_rep->full_elect_timeout; break; case DB_REP_LEASE_TIMEOUT: *timeout = REP_ON(env) ? rep->lease_timeout : db_rep->lease_timeout; break; #ifdef HAVE_REPLICATION_THREADS case DB_REP_ACK_TIMEOUT: *timeout = REP_ON(env) ? rep->ack_timeout : db_rep->ack_timeout; break; case DB_REP_CONNECTION_RETRY: *timeout = REP_ON(env) ? rep->connection_retry_wait : db_rep->connection_retry_wait; break; case DB_REP_ELECTION_RETRY: *timeout = REP_ON(env) ? rep->election_retry_wait : db_rep->election_retry_wait; break; case DB_REP_HEARTBEAT_MONITOR: *timeout = REP_ON(env) ? rep->heartbeat_monitor_timeout : db_rep->heartbeat_monitor_timeout; break; case DB_REP_HEARTBEAT_SEND: *timeout = REP_ON(env) ? rep->heartbeat_frequency : db_rep->heartbeat_frequency; break; #endif default: __db_errx(env, "unknown timeout type argument to DB_ENV->rep_get_timeout"); return (EINVAL); } return (0); } /* * __rep_get_request -- * Get the minimum and maximum number of log records that we wait * before retransmitting. * * PUBLIC: int __rep_get_request * PUBLIC: __P((DB_ENV *, db_timeout_t *, db_timeout_t *)); */ int __rep_get_request(dbenv, minp, maxp) DB_ENV *dbenv; db_timeout_t *minp, *maxp; { DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_get_request", DB_INIT_REP); if (REP_ON(env)) { rep = db_rep->region; ENV_ENTER(env, ip); /* * We acquire the mtx_region or mtx_clientdb mutexes as needed. */ REP_SYSTEM_LOCK(env); if (minp != NULL) DB_TIMESPEC_TO_TIMEOUT((*minp), &rep->request_gap, 0); if (maxp != NULL) DB_TIMESPEC_TO_TIMEOUT((*maxp), &rep->max_gap, 0); REP_SYSTEM_UNLOCK(env); ENV_LEAVE(env, ip); } else { if (minp != NULL) DB_TIMESPEC_TO_TIMEOUT((*minp), &db_rep->request_gap, 0); if (maxp != NULL) DB_TIMESPEC_TO_TIMEOUT((*maxp), &db_rep->max_gap, 0); } return (0); } /* * __rep_set_request -- * Set the minimum and maximum number of log records that we wait * before retransmitting. * * PUBLIC: int __rep_set_request __P((DB_ENV *, db_timeout_t, db_timeout_t)); */ int __rep_set_request(dbenv, min, max) DB_ENV *dbenv; db_timeout_t min, max; { DB_LOG *dblp; DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; LOG *lp; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_request", DB_INIT_REP); if (min == 0 || max < min) { __db_errx(env, "DB_ENV->rep_set_request: Invalid min or max values"); return (EINVAL); } if (REP_ON(env)) { rep = db_rep->region; ENV_ENTER(env, ip); /* * We acquire the mtx_region or mtx_clientdb mutexes as needed. */ REP_SYSTEM_LOCK(env); DB_TIMEOUT_TO_TIMESPEC(min, &rep->request_gap); DB_TIMEOUT_TO_TIMESPEC(max, &rep->max_gap); REP_SYSTEM_UNLOCK(env); MUTEX_LOCK(env, rep->mtx_clientdb); dblp = env->lg_handle; if (dblp != NULL && (lp = dblp->reginfo.primary) != NULL) { DB_TIMEOUT_TO_TIMESPEC(min, &lp->wait_ts); } MUTEX_UNLOCK(env, rep->mtx_clientdb); ENV_LEAVE(env, ip); } else { DB_TIMEOUT_TO_TIMESPEC(min, &db_rep->request_gap); DB_TIMEOUT_TO_TIMESPEC(max, &db_rep->max_gap); } return (0); } /* * __rep_set_transport_pp -- * Set the transport function for replication. * * PUBLIC: int __rep_set_transport_pp __P((DB_ENV *, int, * PUBLIC: int (*)(DB_ENV *, const DBT *, const DBT *, const DB_LSN *, * PUBLIC: int, u_int32_t))); */ int __rep_set_transport_pp(dbenv, eid, f_send) DB_ENV *dbenv; int eid; int (*f_send) __P((DB_ENV *, const DBT *, const DBT *, const DB_LSN *, int, u_int32_t)); { DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; int ret; env = dbenv->env; db_rep = env->rep_handle; ret = 0; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_transport", DB_INIT_REP); if (APP_IS_REPMGR(env)) { __db_errx(env, "DB_ENV->rep_set_transport: cannot call from Replication Manager application"); return (EINVAL); } if (f_send == NULL) { __db_errx(env, "DB_ENV->rep_set_transport: no send function specified"); return (EINVAL); } if (eid < 0) { __db_errx(env, "DB_ENV->rep_set_transport: eid must be greater than or equal to 0"); return (EINVAL); } if ((ret = __rep_set_transport_int(env, eid, f_send)) == 0) /* * Setting a non-repmgr send function makes this a base API * application. */ APP_SET_BASEAPI(env); return (ret); } /* * __rep_set_transport_int -- * Set the internal values for the transport function for replication. * * PUBLIC: int __rep_set_transport_int __P((ENV *, int, * PUBLIC: int (*)(DB_ENV *, const DBT *, const DBT *, const DB_LSN *, * PUBLIC: int, u_int32_t))); */ int __rep_set_transport_int(env, eid, f_send) ENV *env; int eid; int (*f_send) __P((DB_ENV *, const DBT *, const DBT *, const DB_LSN *, int, u_int32_t)); { DB_REP *db_rep; REP *rep; db_rep = env->rep_handle; db_rep->send = f_send; if (REP_ON(env)) { rep = db_rep->region; rep->eid = eid; } else db_rep->eid = eid; return (0); } /* * PUBLIC: int __rep_get_clockskew __P((DB_ENV *, u_int32_t *, u_int32_t *)); */ int __rep_get_clockskew(dbenv, fast_clockp, slow_clockp) DB_ENV *dbenv; u_int32_t *fast_clockp, *slow_clockp; { DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; REP *rep; env = dbenv->env; db_rep = env->rep_handle; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_get_clockskew", DB_INIT_REP); if (REP_ON(env)) { rep = db_rep->region; ENV_ENTER(env, ip); REP_SYSTEM_LOCK(env); *fast_clockp = rep->clock_skew; *slow_clockp = rep->clock_base; REP_SYSTEM_UNLOCK(env); ENV_LEAVE(env, ip); } else { *fast_clockp = db_rep->clock_skew; *slow_clockp = db_rep->clock_base; } return (0); } /* * PUBLIC: int __rep_set_clockskew __P((DB_ENV *, u_int32_t, u_int32_t)); */ int __rep_set_clockskew(dbenv, fast_clock, slow_clock) DB_ENV *dbenv; u_int32_t fast_clock, slow_clock; { DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; REP *rep; int ret; env = dbenv->env; db_rep = env->rep_handle; ret = 0; ENV_NOT_CONFIGURED( env, db_rep->region, "DB_ENV->rep_set_clockskew", DB_INIT_REP); /* * Check for valid values. The fast clock should be a larger * number than the slow clock. We use the slow clock value as * our base for adjustment - therefore, a 2% difference should * be fast == 102, slow == 100. Check for values being 0. If * they are, then set them both to 1 internally. * * We will use these numbers to compute the larger ratio to be * most conservative about the user's intention. */ if (fast_clock == 0 || slow_clock == 0) { /* * If one value is zero, reject if both aren't zero. */ if (slow_clock != 0 || fast_clock != 0) { __db_errx(env, "DB_ENV->rep_set_clockskew: Zero only valid for when used for both arguments"); return (EINVAL); } fast_clock = 1; slow_clock = 1; } if (fast_clock < slow_clock) { __db_errx(env, "DB_ENV->rep_set_clockskew: slow_clock value is larger than fast_clock_value"); return (EINVAL); } if (REP_ON(env)) { rep = db_rep->region; if (IS_REP_STARTED(env)) { __db_errx(env, "DB_ENV->rep_set_clockskew: must be called before DB_ENV->rep_start"); return (EINVAL); } ENV_ENTER(env, ip); REP_SYSTEM_LOCK(env); rep->clock_skew = fast_clock; rep->clock_base = slow_clock; REP_SYSTEM_UNLOCK(env); ENV_LEAVE(env, ip); } else { db_rep->clock_skew = fast_clock; db_rep->clock_base = slow_clock; } return (ret); } /* * __rep_flush -- * Re-push the last log record to all clients, in case they've lost * messages and don't know it. * * PUBLIC: int __rep_flush __P((DB_ENV *)); */ int __rep_flush(dbenv) DB_ENV *dbenv; { DBT rec; DB_LOGC *logc; DB_LSN lsn; DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; int ret, t_ret; env = dbenv->env; db_rep = env->rep_handle; ENV_REQUIRES_CONFIG_XX( env, rep_handle, "DB_ENV->rep_flush", DB_INIT_REP); if (IS_REP_CLIENT(env)) return (0); /* We need a transport function because we send messages. */ if (db_rep->send == NULL) { __db_errx(env, "DB_ENV->rep_flush: must be called after DB_ENV->rep_set_transport"); return (EINVAL); } ENV_ENTER(env, ip); if ((ret = __log_cursor(env, &logc)) != 0) return (ret); memset(&rec, 0, sizeof(rec)); memset(&lsn, 0, sizeof(lsn)); if ((ret = __logc_get(logc, &lsn, &rec, DB_LAST)) != 0) goto err; (void)__rep_send_message(env, DB_EID_BROADCAST, REP_LOG, &lsn, &rec, 0, 0); err: if ((t_ret = __logc_close(logc)) != 0 && ret == 0) ret = t_ret; ENV_LEAVE(env, ip); return (ret); } /* * __rep_sync -- * Force a synchronization to occur between this client and the master. * This is the other half of configuring DELAYCLIENT. * * PUBLIC: int __rep_sync __P((DB_ENV *, u_int32_t)); */ int __rep_sync(dbenv, flags) DB_ENV *dbenv; u_int32_t flags; { DB_LOG *dblp; DB_LSN lsn; DB_REP *db_rep; DB_THREAD_INFO *ip; ENV *env; LOG *lp; REP *rep; int master, ret; u_int32_t repflags, type; env = dbenv->env; db_rep = env->rep_handle; COMPQUIET(flags, 0); ENV_REQUIRES_CONFIG_XX( env, rep_handle, "DB_ENV->rep_sync", DB_INIT_REP); /* We need a transport function because we send messages. */ if (db_rep->send == NULL) { __db_errx(env, "DB_ENV->rep_sync: must be called after DB_ENV->rep_set_transport"); return (EINVAL); } dblp = env->lg_handle; lp = dblp->reginfo.primary; rep = db_rep->region; ret = 0; ENV_ENTER(env, ip); /* * Simple cases. If we're not in the DELAY state we have nothing * to do. If we don't know who the master is, send a MASTER_REQ. */ MUTEX_LOCK(env, rep->mtx_clientdb); lsn = lp->verify_lsn; MUTEX_UNLOCK(env, rep->mtx_clientdb); REP_SYSTEM_LOCK(env); master = rep->master_id; if (master == DB_EID_INVALID) { REP_SYSTEM_UNLOCK(env); (void)__rep_send_message(env, DB_EID_BROADCAST, REP_MASTER_REQ, NULL, NULL, 0, 0); goto out; } /* * We want to hold the rep mutex to test and then clear the * DELAY flag. Racing threads in here could otherwise result * in dual data streams. */ if (!F_ISSET(rep, REP_F_DELAY)) { REP_SYSTEM_UNLOCK(env); goto out; } DB_ASSERT(env, !IS_USING_LEASES(env) || __rep_islease_granted(env) == 0); /* * If we get here, we clear the delay flag and kick off a * synchronization. From this point forward, we will * synchronize until the next time the master changes. */ F_CLR(rep, REP_F_DELAY); if (IS_ZERO_LSN(lsn) && !FLD_ISSET(rep->config, REP_C_AUTOINIT)) { FLD_CLR(rep->lockout_flags, REP_LOCKOUT_ARCHIVE); CLR_RECOVERY_SETTINGS(rep); ret = DB_REP_JOIN_FAILURE; REP_SYSTEM_UNLOCK(env); goto out; } REP_SYSTEM_UNLOCK(env); /* * When we set REP_F_DELAY, we set verify_lsn to the real verify lsn if * we need to verify, or we zeroed it out if this is a client that needs * internal init. So, send the type of message now that * __rep_new_master delayed sending. */ if (IS_ZERO_LSN(lsn)) { DB_ASSERT(env, rep->sync_state == SYNC_UPDATE); type = REP_UPDATE_REQ; repflags = 0; } else { DB_ASSERT(env, rep->sync_state == SYNC_VERIFY); type = REP_VERIFY_REQ; repflags = DB_REP_ANYWHERE; } (void)__rep_send_message(env, master, type, &lsn, NULL, 0, repflags); out: ENV_LEAVE(env, ip); return (ret); } /* * PUBLIC: int __rep_txn_applied __P((ENV *, * PUBLIC: DB_THREAD_INFO *, DB_COMMIT_INFO *, db_timeout_t)); */ int __rep_txn_applied(env, ip, commit_info, timeout) ENV *env; DB_THREAD_INFO *ip; DB_COMMIT_INFO *commit_info; db_timeout_t timeout; { REP *rep; db_timespec limit, now, t; db_timeout_t duration; struct rep_waitgoal reason; int locked, ret, t_ret; if (commit_info->gen == 0) { __db_errx(env, "non-replication commit token in replication env"); return (EINVAL); } rep = env->rep_handle->region; VPRINT(env, (env, DB_VERB_REP_MISC, "checking txn_applied: gen %lu, envid %lu, LSN [%lu][%lu]", (u_long)commit_info->gen, (u_long)commit_info->envid, (u_long)commit_info->lsn.file, (u_long)commit_info->lsn.offset)); locked = 0; __os_gettime(env, &limit, 1); TIMESPEC_ADD_DB_TIMEOUT(&limit, timeout); retry: /* * The checking is done within the scope of the handle count, but if we * end up having to wait that part is not. If a lockout sequence begins * while we're waiting, it will wake us up, and we'll come back here to * try entering the scope again, at which point we'll get an error so * that we return immediately. */ if ((ret = __op_handle_enter(env)) != 0) goto out; ret = __rep_check_applied(env, ip, commit_info, &reason); t_ret = __env_db_rep_exit(env); /* * Between here and __rep_check_applied() we use DB_TIMEOUT privately to * mean that the transaction hasn't been applied yet, but it still * plausibly could be soon; think of it as meaning "not yet". So * DB_TIMEOUT doesn't necessarily mean that DB_TIMEOUT is the ultimate * return that the application will see. * * When we get this "not yet", we check the actual time remaining. If * the time has expired, then indeed we can simply pass DB_TIMEOUT back * up to the calling application. But if not, it tells us that we have * a chance to wait and try again. This is a nice division of labor, * because it means the lower level functions (__rep_check_applied() and * below) do not have to mess with any actual time computations, or * waiting, at all. */ if (ret == DB_TIMEOUT && t_ret == 0 && F_ISSET(rep, REP_F_CLIENT)) { __os_gettime(env, &now, 1); if (timespeccmp(&now, &limit, <)) { /* Compute how much time remains before the limit. */ t = limit; timespecsub(&t, &now); DB_TIMESPEC_TO_TIMEOUT(duration, &t, 1); /* * Wait for whatever __rep_check_applied told us we * needed to wait for. But first, check the condition * again under mutex protection, in case there was a * close race. */ if (reason.why == AWAIT_LSN || reason.why == AWAIT_HISTORY) { MUTEX_LOCK(env, rep->mtx_clientdb); locked = 1; } REP_SYSTEM_LOCK(env); ret = __rep_check_goal(env, &reason); if (locked) { MUTEX_UNLOCK(env, rep->mtx_clientdb); locked = 0; } if (ret == DB_TIMEOUT) { /* * The usual case: we haven't reached our goal * yet, even after checking again while holding * mutex. */ ret = __rep_await_condition(env, &reason, duration); /* * If it were possible for * __rep_await_condition() to return DB_TIMEOUT * that would confuse the outer "if" statement * here. */ DB_ASSERT(env, ret != DB_TIMEOUT); } REP_SYSTEM_UNLOCK(env); if (ret != 0) goto out; /* * Note that the "reason" that check_applied set, and * that await_condition waited for, does not necessarily * represent a final result ready to return to the * user. In some cases there may be a few state changes * necessary before we are able to determine the final * result. Thus whenever we complete a successful wait * we need to cycle back and check the full txn_applied * question again. */ goto retry; } } if (t_ret != 0 && (ret == 0 || ret == DB_TIMEOUT || ret == DB_NOTFOUND)) ret = t_ret; out: return (ret); } /* * The only non-zero return code from this function is for unexpected errors. * We normally return 0, regardless of whether the wait terminated because the * condition was satisfied or the timeout expired. */ static int __rep_await_condition(env, reasonp, duration) ENV *env; struct rep_waitgoal *reasonp; db_timeout_t duration; { REGENV *renv; REGINFO *infop; REP *rep; struct __rep_waiter *waiter; int ret; rep = env->rep_handle->region; infop = env->reginfo; renv = infop->primary; /* * Acquire the first lock on the self-blocking mutex when we first * allocate it. Thereafter when it's on the free list we know that * first lock has already been taken. */ if ((waiter = SH_TAILQ_FIRST(&rep->free_waiters, __rep_waiter)) == NULL) { MUTEX_LOCK(env, renv->mtx_regenv); if ((ret = __env_alloc(env->reginfo, sizeof(struct __rep_waiter), &waiter)) == 0) { memset(waiter, 0, sizeof(*waiter)); if ((ret = __mutex_alloc(env, MTX_REP_WAITER, DB_MUTEX_SELF_BLOCK, &waiter->mtx_repwait)) != 0) __env_alloc_free(infop, waiter); } MUTEX_UNLOCK(env, renv->mtx_regenv); if (ret != 0) return (ret); MUTEX_LOCK(env, waiter->mtx_repwait); } else SH_TAILQ_REMOVE(&rep->free_waiters, waiter, links, __rep_waiter); waiter->flags = 0; waiter->goal = *reasonp; SH_TAILQ_INSERT_HEAD(&rep->waiters, waiter, links, __rep_waiter); VPRINT(env, (env, DB_VERB_REP_MISC, "waiting for condition %d", (int)reasonp->why)); REP_SYSTEM_UNLOCK(env); /* Wait here for conditions to become more favorable. */ MUTEX_WAIT(env, waiter->mtx_repwait, duration); REP_SYSTEM_LOCK(env); if (!F_ISSET(waiter, REP_F_WOKEN)) SH_TAILQ_REMOVE(&rep->waiters, waiter, links, __rep_waiter); SH_TAILQ_INSERT_HEAD(&rep->free_waiters, waiter, links, __rep_waiter); return (0); } /* * Check whether the transaction is currently applied. If it is not, but it * might likely become applied in the future, then return DB_TIMEOUT. It's the * caller's duty to figure out whether to wait or not in that case. Here we * only do an immediate check of the current state of affairs. */ static int __rep_check_applied(env, ip, commit_info, reasonp) ENV *env; DB_THREAD_INFO *ip; DB_COMMIT_INFO *commit_info; struct rep_waitgoal *reasonp; { DB_LOG *dblp; DB_REP *db_rep; LOG *lp; REP *rep; DB_TXN *txn; DBC *dbc; __rep_lsn_hist_data_args hist, hist2; DB_LSN lsn; u_int32_t gen; int ret, t_ret; db_rep = env->rep_handle; rep = db_rep->region; dblp = env->lg_handle; lp = dblp->reginfo.primary; gen = rep->gen; txn = NULL; dbc = NULL; if (F_ISSET(rep, REP_F_MASTER)) { LOG_SYSTEM_LOCK(env); lsn = lp->lsn; LOG_SYSTEM_UNLOCK(env); } else { MUTEX_LOCK(env, rep->mtx_clientdb); lsn = lp->max_perm_lsn; MUTEX_UNLOCK(env, rep->mtx_clientdb); } /* * The first thing to consider is whether we're in the right gen. * The token gen either matches our current gen, or is left over from an * older gen, or in rare circumstances could be from a "future" gen that * we haven't learned about yet (or that got rolled back). */ if (commit_info->gen == gen) { ret = __rep_read_lsn_history(env, ip, &txn, &dbc, gen, &hist, reasonp, DB_SET); if (ret == DB_NOTFOUND) { /* * We haven't yet received the LSN history of the * current generation from the master. Return * DB_TIMEOUT to tell the caller it needs to wait and * tell it to wait for the LSN history. * * Note that this also helps by eliminating the weird * period between receiving a new gen (from a NEWMASTER) * and the subsequent syncing with that new gen. We * really only want to return success at the current gen * once we've synced. */ ret = DB_TIMEOUT; reasonp->why = AWAIT_HISTORY; reasonp->u.lsn = lsn; } if (ret != 0) goto out; if (commit_info->envid != hist.envid) { /* * Gens match, but envids don't: means there were two * masters at the same gen, and the txn of interest was * rolled back. */ ret = DB_NOTFOUND; goto out; } if (LOG_COMPARE(&commit_info->lsn, &lsn) > 0) { /* * We haven't yet gotten the LSN of interest, but we can * expect it soon; so wait for it. */ ret = DB_TIMEOUT; reasonp->why = AWAIT_LSN; reasonp->u.lsn = commit_info->lsn; goto out; } if (LOG_COMPARE(&commit_info->lsn, &hist.lsn) >= 0) { /* * The LSN of interest is in the past, but within the * range claimed for this gen. Success! (We have read * consistency.) */ ret = 0; goto out; } /* * There must have been a DUPMASTER at some point: the * description of the txn of interest doesn't match what we see * in the history available to us now. */ ret = DB_NOTFOUND; } else if (commit_info->gen < gen || gen == 0) { /* * Transaction from an old gen. Read this gen's base LSN, plus * that of the next higher gen, because we want to check that * the token LSN is within the close/open range defined by * [base,next). */ ret = __rep_read_lsn_history(env, ip, &txn, &dbc, commit_info->gen, &hist, reasonp, DB_SET); t_ret = __rep_read_lsn_history(env, ip, &txn, &dbc, commit_info->gen, &hist2, reasonp, DB_NEXT); if (ret == DB_NOTFOUND) { /* * If the desired gen is not in our database, it could * mean either of two things. 1. The whole gen could * have been rolled back. 2. We could just be really * far behind on replication. Reading ahead to the next * following gen, which we likely need anyway, helps us * decide which case to conclude. */ if (t_ret == 0) /* * Second read succeeded, so "being behind in * replication" is not a viable reason for * having failed to find the first read. * Therefore, the gen must have been rolled * back, and the proper result is NOTFOUND to * indicate that. */ goto out; if (t_ret == DB_NOTFOUND) { /* * Second read also got a NOTFOUND: we're * definitely "behind" (we don't even have * current gen's history). So, waiting is the * correct result. */ ret = DB_TIMEOUT; reasonp->why = AWAIT_HISTORY; reasonp->u.lsn = lsn; goto out; } /* * Here, t_ret is something unexpected, which trumps the * NOTFOUND returned from the first read. */ ret = t_ret; goto out; } if (ret != 0) goto out; /* Unexpected error, first read. */ if (commit_info->envid != hist.envid) { /* * (We don't need the second read in order to make this * test.) * * We have info for the indicated gen, but the envids * don't match, meaning the txn was written at a dup * master and that gen instance was rolled back. */ ret = DB_NOTFOUND; goto out; } /* Examine result of second read. */ if ((ret = t_ret) == DB_NOTFOUND) { /* * We haven't even heard about our current gen yet, so * it's worth waiting for it. */ ret = DB_TIMEOUT; reasonp->why = AWAIT_HISTORY; reasonp->u.lsn = lsn; } else if (ret != 0) goto out; /* Second read returned unexpeced error. */ /* * We now have the history info for the gen of the txn, and for * the subsequent gen. All we have to do is see if the LSN is * in range. */ if (LOG_COMPARE(&commit_info->lsn, &hist.lsn) >= 0 && LOG_COMPARE(&commit_info->lsn, &hist2.lsn) < 0) ret = 0; else ret = DB_NOTFOUND; } else { /* * Token names a future gen. If we're a client and the LSN also * is in the future, then it's possible we just haven't caught * up yet, so we can wait for it. Otherwise, it must have been * part of a generation that got lost in a roll-back. */ if (F_ISSET(rep, REP_F_CLIENT) && LOG_COMPARE(&commit_info->lsn, &lsn) > 0) { reasonp->why = AWAIT_GEN; reasonp->u.gen = commit_info->gen; return (DB_TIMEOUT); } return (DB_NOTFOUND); } out: if (dbc != NULL && (t_ret = __dbc_close(dbc)) != 0 && ret == 0) ret = t_ret; if (txn != NULL && (t_ret = __db_txn_auto_resolve(env, txn, 1, ret)) != 0 && ret == 0) ret = t_ret; return (ret); } /* * The txn and dbc handles are owned by caller, though we create them if * necessary. Caller is responsible for closing them. */ static int __rep_read_lsn_history(env, ip, txn, dbc, gen, gen_infop, reasonp, flags) ENV *env; DB_THREAD_INFO *ip; DB_TXN **txn; DBC **dbc; u_int32_t gen; __rep_lsn_hist_data_args *gen_infop; struct rep_waitgoal *reasonp; u_int32_t flags; { DB_REP *db_rep; REP *rep; DB *dbp; __rep_lsn_hist_key_args key; u_int8_t key_buf[__REP_LSN_HIST_KEY_SIZE]; u_int8_t data_buf[__REP_LSN_HIST_DATA_SIZE]; DBT key_dbt, data_dbt; u_int32_t desired_gen; int ret, tries; db_rep = env->rep_handle; rep = db_rep->region; ret = 0; DB_ASSERT(env, flags == DB_SET || flags == DB_NEXT); /* Simply return cached info, if we already have it. */ desired_gen = flags == DB_SET ? gen : gen + 1; REP_SYSTEM_LOCK(env); if (rep->gen == desired_gen && !IS_ZERO_LSN(rep->gen_base_lsn)) { gen_infop->lsn = rep->gen_base_lsn; gen_infop->envid = rep->master_envid; goto unlock; } REP_SYSTEM_UNLOCK(env); tries = 0; retry: if (*txn == NULL && (ret = __txn_begin(env, ip, NULL, txn, 0)) != 0) return (ret); if ((dbp = db_rep->lsn_db) == NULL) { if ((ret = __rep_open_lsn_history(env, ip, *txn, 0, &dbp)) != 0) { /* * If the database isn't there, it could be because it's * memory-resident, and we haven't yet sync'ed with the * master to materialize it. (It could make sense to * include a test for INMEM in this conditional * expression, if we were sure all sites had matching * INMEM settings; but since we don't enforce that, * leaving it out makes for more optimistic behavior.) */ if (ret == ENOENT && !F_ISSET(rep, REP_F_NIMDBS_LOADED | REP_F_MASTER)) { ret = DB_TIMEOUT; reasonp->why = AWAIT_NIMDB; } goto err; } db_rep->lsn_db = dbp; } if (*dbc == NULL && (ret = __db_cursor(dbp, ip, *txn, dbc, 0)) != 0) goto err; if (flags == DB_SET) { key.version = REP_LSN_HISTORY_FMT_VERSION; key.gen = gen; __rep_lsn_hist_key_marshal(env, &key, key_buf); } DB_INIT_DBT(key_dbt, key_buf, __REP_LSN_HIST_KEY_SIZE); key_dbt.ulen = __REP_LSN_HIST_KEY_SIZE; F_SET(&key_dbt, DB_DBT_USERMEM); memset(&data_dbt, 0, sizeof(data_dbt)); data_dbt.data = data_buf; data_dbt.ulen = __REP_LSN_HIST_DATA_SIZE; F_SET(&data_dbt, DB_DBT_USERMEM); if ((ret = __dbc_get(*dbc, &key_dbt, &data_dbt, flags)) != 0) { if ((ret == DB_LOCK_DEADLOCK || ret == DB_LOCK_NOTGRANTED) && ++tries < 5) { /* Limit of 5 is an arbitrary choice. */ ret = __dbc_close(*dbc); *dbc = NULL; if (ret != 0) goto err; ret = __txn_abort(*txn); *txn = NULL; if (ret != 0) goto err; __os_yield(env, 0, 10000); /* Arbitrary duration. */ goto retry; } goto err; } /* * In the DB_NEXT case, we don't know what the next gen is. Unmarshal * the key too, just so that we can check whether it matches the current * gen, for setting the cache. Note that, interestingly, the caller * doesn't care what the key is in that case! */ if ((ret = __rep_lsn_hist_key_unmarshal(env, &key, key_buf, __REP_LSN_HIST_KEY_SIZE, NULL)) != 0) goto err; ret = __rep_lsn_hist_data_unmarshal(env, gen_infop, data_buf, __REP_LSN_HIST_DATA_SIZE, NULL); REP_SYSTEM_LOCK(env); if (rep->gen == key.gen) { rep->gen_base_lsn = gen_infop->lsn; rep->master_envid = gen_infop->envid; } unlock: REP_SYSTEM_UNLOCK(env); err: return (ret); } /* * __rep_conv_vers -- * Convert from a log version to the replication message version * that release used. */ static u_int32_t __rep_conv_vers(env, log_ver) ENV *env; u_int32_t log_ver; { COMPQUIET(env, NULL); /* * We can't use a switch statement, some of the DB_LOGVERSION_XX * constants are the same */ if (log_ver == DB_LOGVERSION) return (DB_REPVERSION); if (log_ver == DB_LOGVERSION_48p2) return (DB_REPVERSION_48); if (log_ver == DB_LOGVERSION_48) return (DB_REPVERSION_48); if (log_ver == DB_LOGVERSION_47) return (DB_REPVERSION_47); if (log_ver == DB_LOGVERSION_46) return (DB_REPVERSION_46); if (log_ver == DB_LOGVERSION_45) return (DB_REPVERSION_45); if (log_ver == DB_LOGVERSION_44) return (DB_REPVERSION_44); return (DB_REPVERSION_INVALID); }